Merge pull request #52 from wirepair/ordered_events

Ordered events

CHANGELOG.md

@@ -1,4 +1,5 @@
 # Changelog (2021)
+- 2.2.0 (July 21st) Dispatching DevTool events via a newly spawned go routine was causing messages to be delivered out of order. This change synchronizes them using an internal channel. Note: If you previously had Subscriptions try to signal each other (via another channel) it may be blocked since all subscriptions are executed under a single go routine now. Upgrade with caution.
 - 2.1.6 (June 9th) Fix go routine leak and add test
 - 2.1.5 (June 8th) Fix race condition on error and endpoint
 - 2.1.4 (June 8th) Added a chrome exit handler from @camswords.

v2/chrome_target.go

@@ -49,6 +49,12 @@ type TargetInfo struct {
 	WebSocketDebuggerUrl string `json:"webSocketDebuggerUrl"`
 }
 
+// devtoolsEventResponse from websocket with method and msg data to dispatch to subscribers
+type devtoolsEventResponse struct {
+	Method string `json:"method"`
+	Msg    []byte `json:"msg"`
+}
+
 // ChromeTarget (Tab/Process). Messages are returned to callers via non-buffered channels. Helpfully,
 // the remote debugger service uses id's so we can correlate which request should match which response.
 // We use a map to store the id of the request which contains a reference to a gcdmessage.Message that holds the
@@ -113,10 +119,11 @@ type ChromeTarget struct {
 	WebAudio             *gcdapi.WebAudio
 	WebAuthn             *gcdapi.WebAuthn
 
-	Target          *TargetInfo              // The target information see, TargetInfo
-	sendCh          chan *gcdmessage.Message // The channel used for API components to send back to use
-	doneCh          chan struct{}            // we be donez.
-	apiTimeout      time.Duration            // A customizable timeout for waiting on Chrome to respond to us
+	Target          *TargetInfo                 // The target information see, TargetInfo
+	sendCh          chan *gcdmessage.Message    // The channel used for API components to send back to use
+	eventCh         chan *devtoolsEventResponse // The channel used for dispatching devtool events
+	doneCh          chan struct{}               // we be donez.
+	apiTimeout      time.Duration               // A customizable timeout for waiting on Chrome to respond to us
 	logger          Log
 	debugger        *Gcd
 	stopped         bool // we are/have shutdown
@@ -138,6 +145,7 @@ func openChromeTarget(debugger *Gcd, target *TargetInfo, observer observer.Messa
 		sendCh:          make(chan *gcdmessage.Message),
 		replyDispatcher: make(map[int64]chan *gcdmessage.Message),
 		eventDispatcher: make(map[string]func(*ChromeTarget, []byte)),
+		eventCh:         make(chan *devtoolsEventResponse, debugger.eventQueueSize), // allow enough events to buffer up
 		doneCh:          make(chan struct{}),
 		logger:          debugger.logger,
 		debugger:        debugger,
@@ -252,6 +260,7 @@ func (c *ChromeTarget) Unsubscribe(method string) {
 func (c *ChromeTarget) listen() {
 	go c.listenRead()
 	go c.listenWrite()
+	go c.dispatchEvents()
 }
 
 // Listens for API components wishing to send requests to the Chrome Debugger Service
@@ -321,20 +330,39 @@ func (c *ChromeTarget) dispatchResponse(msg []byte) {
 	c.checkTargetDisconnected(f.Method)
 
 	c.eventLock.RLock()
-	if r, ok := c.eventDispatcher[f.Method]; ok {
-		c.eventLock.RUnlock()
-		c.logDebug("dispatching", f.Method, "event: ", string(msg))
-		go r(c, msg)
-		return
+	_, ok := c.eventDispatcher[f.Method]
+	c.eventLock.RUnlock()
 
+	if ok {
+		c.eventCh <- &devtoolsEventResponse{Method: f.Method, Msg: msg}
+		return
 	}
-	c.eventLock.RUnlock()
 
 	if c.debugger.debugEvents {
 		c.logger.Println("no event reciever bound: ", f.Method, " data: ", string(msg))
 	}
 }
 
+func (c *ChromeTarget) dispatchEvents() {
+	for {
+		select {
+		case <-c.doneCh:
+			return
+		case <-c.ctx.Done():
+			return
+		case m := <-c.eventCh:
+			c.logDebug("dispatching", m.Method, "event: ", string(m.Msg))
+			c.eventLock.Lock()
+			cb, ok := c.eventDispatcher[m.Method]
+			c.eventLock.Unlock()
+			if !ok {
+				break
+			}
+			cb(c, m.Msg)
+		}
+	}
+}
+
 func (c *ChromeTarget) dispatchWithTimeout(r chan<- *gcdmessage.Message, id int64, msg []byte) {
 	timeout := time.NewTimer(c.GetApiTimeout())
 	defer timeout.Stop()

v2/gcd.go

@@ -44,7 +44,7 @@ import (
 
 var json = jsoniter.ConfigCompatibleWithStandardLibrary
 
-var GCDVERSION = "v2.1.6"
+var GCDVERSION = "v2.2.0"
 
 var (
 	ErrNoTabAvailable = errors.New("no available tab found")
@@ -90,6 +90,7 @@ type Gcd struct {
 	flags               []string
 	env                 []string
 	chomeApiVersion     string
+	eventQueueSize      int
 	ctx                 context.Context
 	logger              Log
 	debugEvents         bool
@@ -107,6 +108,7 @@ func NewChromeDebugger(opts ...func(*Gcd)) *Gcd {
 	c.onChromeExitHandler = nil
 	c.flags = make([]string, 0)
 	c.env = make([]string, 0)
+	c.eventQueueSize = 256
 	c.ctx = context.Background()
 	c.logger = LogDiscarder{}
 	c.messageObserver = observer.NewIgnoreMessagesObserver()
@@ -117,6 +119,13 @@ func NewChromeDebugger(opts ...func(*Gcd)) *Gcd {
 	return c
 }
 
+// WithEventQueueSize number of DevTool events to allow to queue up
+func WithEventQueueSize(queueSize int) func(*Gcd) {
+	return func(g *Gcd) {
+		g.eventQueueSize = queueSize
+	}
+}
+
 // WithTerminationHandler Pass a handler to be notified when the chrome process exits.
 func WithTerminationHandler(handler TerminatedHandler) func(*Gcd) {
 	return func(g *Gcd) {