fix(deps): update github.com/digitalocean/go-qemu digest to f035778

Signed-off-by: Renovate Bot <bot@renovateapp.com>

vendor/github.com/digitalocean/go-libvirt/internal/constants/remote_protocol.gen.go

@@ -881,6 +881,8 @@ const (
 	ProcDomainAuthorizedSshKeysGet = 424
 	// ProcDomainAuthorizedSshKeysSet is libvirt's REMOTE_PROC_DOMAIN_AUTHORIZED_SSH_KEYS_SET
 	ProcDomainAuthorizedSshKeysSet = 425
+	// ProcDomainGetMessages is libvirt's REMOTE_PROC_DOMAIN_GET_MESSAGES
+	ProcDomainGetMessages = 426
 
 
 	// From consts:
@@ -1010,6 +1012,8 @@ const (
 	NetworkPortParametersMax = 16
 	// DomainAuthorizedSshKeysMax is libvirt's REMOTE_DOMAIN_AUTHORIZED_SSH_KEYS_MAX
 	DomainAuthorizedSshKeysMax = 2048
+	// DomainMessagesMax is libvirt's REMOTE_DOMAIN_MESSAGES_MAX
+	DomainMessagesMax = 2048
 	// DomainEventGraphicsIdentityMax is libvirt's REMOTE_DOMAIN_EVENT_GRAPHICS_IDENTITY_MAX
 	DomainEventGraphicsIdentityMax = 20
 	// Program is libvirt's REMOTE_PROGRAM

vendor/github.com/digitalocean/go-libvirt/internal/event/stream.go

@@ -14,7 +14,16 @@
 
 package event
 
-import "context"
+import (
+	"context"
+)
+
+// emptyEvent is used as a zero-value. Clients will never receive one of these;
+// they are only here to satisfy the compiler. See the comments in process() for
+// more information.
+type emptyEvent struct{}
+
+func (emptyEvent) GetCallbackID() int32 { return 0 }
 
 // Stream is an unbounded buffered event channel. The implementation
 // consists of a pair of unbuffered channels and a goroutine to manage them.
@@ -28,12 +37,45 @@ type Stream struct {
 
 	// manage unbounded channel behavior.
 	queue   []Event
+	qlen    chan (chan int)
 	in, out chan Event
 
 	// terminates processing
 	shutdown context.CancelFunc
 }
 
+// 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 {
+	s := &Stream{
+		Program:    program,
+		CallbackID: cbID,
+		in:         make(chan Event),
+		out:        make(chan Event),
+		qlen:       make(chan (chan int)),
+	}
+
+	// Start the processing loop, which will return a routine we can use to
+	// shut the queue down later.
+	s.shutdown = s.start()
+
+	return s
+}
+
+// Len will return the current count of events in the internal queue for a
+// stream. It does this by sending a message to the stream's process() loop,
+// which will then write the current length to the channel contained in that
+// message.
+func (s *Stream) Len() int {
+	// Send a request to the process() loop to get the current length of the
+	// queue
+	ch := make(chan int)
+	s.qlen <- ch
+	return <-ch
+}
+
 // Recv returns the next available event from the Stream's queue.
 func (s *Stream) Recv() chan Event {
 	return s.out
@@ -44,9 +86,9 @@ 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.
+// 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()
@@ -54,8 +96,7 @@ func (s *Stream) 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.
+// terminated by the returned context.CancelFunc.
 func (s *Stream) start() context.CancelFunc {
 	ctx, cancel := context.WithCancel(context.Background())
 
@@ -64,82 +105,53 @@ func (s *Stream) start() context.CancelFunc {
 	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.
+// 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
+// listening client. New events pushed onto the queue will not block if the
+// client is not actively polling for them; the stream will buffer them
+// internally.
 func (s *Stream) process(ctx context.Context) {
-	defer func() {
-		close(s.in)
-		close(s.out)
-	}()
+	// Close the output channel so that clients know this stream is finished.
+	// We don't close s.in to avoid creating a race with the stream's Push()
+	// function.
+	defer close(s.out)
 
+	// This function is used to retrieve the next event from the queue, to be
+	// sent to the client. If there are no more events to send, it returns a nil
+	// channel and a zero-value event.
+	nextEvent := func() (chan Event, Event) {
+		sendCh := chan Event(nil)
+		next := Event(emptyEvent{})
+		if len(s.queue) > 0 {
+			sendCh = s.out
+			next = s.queue[0]
+		}
+		return sendCh, next
+	}
+
+	// The select statement in this loop relies on the fact that a send to a nil
+	// channel will block forever. If we have no entries in the queue, the
+	// sendCh variable will be nil, so the clause that attempts to send an event
+	// to the client will never complete. Clients will never receive an
+	// emptyEvent.
 	for {
-		// informs send() to stop trying
-		nctx, next := context.WithCancel(ctx)
-		defer next()
+		sendCh, nextEvt := nextEvent()
 
 		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{}
-			}
+		case lenCh := <-s.qlen:
+			lenCh <- len(s.queue)
+
+		// client received an event, pop from queue
+		case sendCh <- nextEvt:
+			s.queue = s.queue[1:]
 
 		// 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
-}