worker_node: reconnect User stream properly

2 files changed, 11 insertions, 6 deletions

diff --git a/worker_node/lib/user_connection.rb b/worker_node/lib/user_connection.rb
index 3d8be15..d8a75ad 100644
--- a/worker_node/lib/user_connection.rb
+++ b/worker_node/lib/user_connection.rb
@@ -21,7 +21,7 @@ class UserConnection
   end
 
   def stop
-    @client.close
+    @client.stop
     log(:info, "Stopped: #{@account_id}")
   end
 
@@ -35,7 +35,7 @@ class UserConnection
         log(:warn, "Connection reset")
         EM.add_timer(5) { @client.reconnect }
       else
-        log(:error, "Unknown error: #{error.inspect}")
+        log(:error, "Unknown error: #{error}")
       end
     end
     @client.on_service_unavailable do |message|
@@ -54,7 +54,8 @@ class UserConnection
       log(:warn, "420: #{message}")
     end
     @client.on_disconnected do
-      @client.reconnect
+      log(:warn, "Disconnected")
+      EM.add_timer(5) { @client.reconnect }
     end
 
     @client.on_item do |item|
diff --git a/worker_node/lib/user_stream/client.rb b/worker_node/lib/user_stream/client.rb
index 1f83914..4044292 100644
--- a/worker_node/lib/user_stream/client.rb
+++ b/worker_node/lib/user_stream/client.rb
@@ -7,7 +7,7 @@ module UserStream
     def initialize(options = {})
       @options = { compression: true }.merge(options).freeze
       @callbacks = {}
-      @closing = false
+      @exiting = false
     end
 
     def update(options = {})
@@ -20,8 +20,12 @@ module UserStream
       connect
     end
 
+    def stop
+      @exiting = true
+      close
+    end
+
     def close
-      @closing = true
       @http.close
     end
 
@@ -64,7 +68,7 @@ module UserStream
       end
 
       http.errback do
-        callback(:error, http.error) unless @closing
+        callback(:error, http.error) unless @exiting
       end
 
       @http = http