From bc26bf9dd9f532f710cbff830f35875d4adca0bb Mon Sep 17 00:00:00 2001
From: Iain Beeston <iainbeeston@users.noreply.github.com>
Date: Tue, 22 Jan 2019 09:00:05 +0000
Subject: [PATCH] Don't discard unwanted messages

Right now we process messages with a queue, and discard any that don't
match the id that we're waiting for. This means that messages can't be
processed out of order, which can happen with a multithreaded
architecture. I've refactored it to use a synchronized hash rather
than a queue, which allows us to retrieve messages in any order.
---
 lib/capybara/cuprite/browser/client.rb | 26 +++++++++++++++++---------
 1 file changed, 17 insertions(+), 9 deletions(-)

diff --git a/lib/capybara/cuprite/browser/client.rb b/lib/capybara/cuprite/browser/client.rb
index afc5141f..b9bc570f 100644
--- a/lib/capybara/cuprite/browser/client.rb
+++ b/lib/capybara/cuprite/browser/client.rb
@@ -6,13 +6,14 @@
 module Capybara::Cuprite
   class Browser
     class Client
-      class IdError < RuntimeError; end
 
       def initialize(browser, ws_url)
         @command_id = 0
         @subscribed = Hash.new { |h, k| h[k] = [] }
         @browser = browser
-        @commands = Queue.new
+        @commands = {}
+        @mutex = Mutex.new
+        @resource = ConditionVariable.new
         @ws = WebSocket.new(ws_url, @browser.logger)
 
         @thread = Thread.new do
@@ -21,11 +22,12 @@ def initialize(browser, ws_url)
             if method
               @subscribed[method].each { |b| b.call(params) }
             else
-              @commands.push(message)
+              @mutex.synchronize do
+                @commands[message["id"]] = message
+                @resource.broadcast
+              end
             end
           end
-
-          @commands.close
         end
       end
 
@@ -36,14 +38,20 @@ def command(method, params = {})
       end
 
       def wait(id:)
-        message = Timeout.timeout(@browser.timeout, TimeoutError) { @commands.pop }
+        message = nil
+        Timeout.timeout(@browser.timeout, TimeoutError) do
+          message = @mutex.synchronize { @commands.delete(id) }
+          while !message
+            message = @mutex.synchronize do
+              @resource.wait(@mutex)
+              @commands.delete(id)
+            end
+          end
+        end
         raise DeadBrowser unless message
-        raise IdError if message["id"] != id
         error, response = message.values_at("error", "result")
         raise BrowserError.new(error) if error
         response
-      rescue IdError
-        retry
       end
 
       def subscribe(event, &block)