1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
// Copyright 2020 The go-libvirt Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package event
import "context"
// Stream is an unbounded buffered event channel. The implementation
// consists of a pair of unbuffered channels and a goroutine to manage them.
// Client behavior will not cause incoming events to block.
type Stream struct {
// Program specifies the source of the events - libvirt or QEMU.
Program uint32
// CallbackID is returned by the event registration call.
CallbackID int32
// manage unbounded channel behavior.
queue []Event
in, out chan Event
// terminates processing
shutdown context.CancelFunc
}
// Recv returns the next available event from the Stream's queue.
func (s *Stream) Recv() chan Event {
return s.out
}
// Push appends a new event to the queue.
func (s *Stream) Push(e Event) {
s.in <- e
}
// Shutdown gracefully terminates Stream processing, releasing all
// internal resources. Events which have not yet been received by the client
// will be dropped. Subsequent calls to Shutdown() are idempotent.
func (s *Stream) Shutdown() {
if s.shutdown != nil {
s.shutdown()
}
}
// start starts the event processing loop, which will continue to run until
// terminated by the returned context.CancelFunc. Starting a previously started
// Stream is an idempotent operation.
func (s *Stream) start() context.CancelFunc {
ctx, cancel := context.WithCancel(context.Background())
go s.process(ctx)
return cancel
}
// process manages an Stream's lifecycle until canceled by the provided
// context. Incoming events are appended to a queue which is then relayed to
// the a listening client. New events pushed onto the queue will not block due
// to client behavior.
func (s *Stream) process(ctx context.Context) {
defer func() {
close(s.in)
close(s.out)
}()
for {
// informs send() to stop trying
nctx, next := context.WithCancel(ctx)
defer next()
select {
// new event received, append to queue
case e := <-s.in:
s.queue = append(s.queue, e)
// client recieved an event, pop from queue
case <-s.send(nctx):
if len(s.queue) > 1 {
s.queue = s.queue[1:]
} else {
s.queue = []Event{}
}
// shutdown requested
case <-ctx.Done():
return
}
next()
}
}
// send returns a channel which blocks until either the first item on the queue
// (if existing) is sent to the client, or the provided context is canceled.
// The stream's queue is never modified.
func (s *Stream) send(ctx context.Context) <-chan struct{} {
ch := make(chan struct{})
go func() {
defer close(ch)
// do nothing and block if the queue is empty
if len(s.queue) == 0 {
<-ctx.Done()
return
}
// otherwise, attempt to send the event
select {
case s.out <- s.queue[0]:
case <-ctx.Done():
}
}()
return ch
}
// NewStream configures a new Event Stream. Incoming events are appended to a
// queue, which is then relayed to the listening client. Client behavior will
// not cause incoming events to block. It is the responsibility of the caller
// to terminate the Stream via Shutdown() when no longer in use.
func NewStream(program uint32, cbID int32) *Stream {
ic := &Stream{
Program: program,
CallbackID: cbID,
in: make(chan Event),
out: make(chan Event),
}
ic.shutdown = ic.start()
return ic
}
|
