Replace websocket-client-simple gem with async-websocket

Gemfile

@@ -10,7 +10,6 @@ gem "digest-crc"
 gem "ircparser"
 gem "rexml"
 gem "rubyzip"
-gem "websocket-client-simple"
 gem "win32-process", platforms: [:windows]
 gem "win32-security", platforms: [:windows]
 

lib/api/server_list_updater.rb

@@ -2,6 +2,9 @@ class W3DHub
   class Api
     class ServerListUpdater
       LOG_TAG = "W3DHub::Api::ServerListUpdater".freeze
+
+      TYPE_PING = 6
+
       include CyberarmEngine::Common
       @@instance = nil
 
@@ -15,6 +18,7 @@ class W3DHub
 
       def initialize
         @auto_reconnect = false
+        @reconnection_delay = 1
 
         @invocation_id = 0
 
@@ -23,16 +27,14 @@ class W3DHub
       end
 
       def run
-        return
-
         Thread.new do
           Sync do |task|
             begin
-              async_connect(task)
+              @auto_reconnect = true
 
               while W3DHub::BackgroundWorker.alive?
-                async_connect(task) if @auto_reconnect
-                sleep 1
+                connect if @auto_reconnect
+                sleep @reconnection_delay
               end
             rescue => e
               puts e
@@ -48,39 +50,6 @@ class W3DHub
         @@instance = nil
       end
 
-      def async_connect(task)
-        @auto_reconnect = false
-
-        logger.debug(LOG_TAG) { "Requesting connection token..." }
-        response = Api.post("/listings/push/v2/negotiate?negotiateVersion=1", Api::DEFAULT_HEADERS, "", :gsh)
-
-        if response.status != 200
-          @auto_reconnect = true
-          return
-        end
-
-        data = JSON.parse(response.body, symbolize_names: true)
-
-        @invocation_id = 0 if @invocation_id > 9095
-        id = data[:connectionToken]
-        endpoint = "#{Api::SERVER_LIST_ENDPOINT}/listings/push/v2?id=#{id}"
-
-        logger.debug(LOG_TAG) { "Connecting to websocket..." }
-
-        Async::WebSocket::Client.connect(Async::HTTP::Endpoint.parse(endpoint)) do |connection|
-          logger.debug(LOG_TAG) { "Requesting json protocol, v1..." }
-          async_websocket_send(connection, { protocol: "json", version: 1 }.to_json)
-        end
-      end
-
-      def async_websocket_send(connection, payload)
-        connection.write("#{payload}\x1e")
-        connection.flush
-      end
-
-      def async_websocket_read(connection, payload)
-      end
-
       def connect
         @auto_reconnect = false
 
@@ -89,9 +58,13 @@ class W3DHub
 
         if response.status != 200
           @auto_reconnect = true
+          @reconnection_delay = @reconnection_delay * 2
+          @reconnection_delay = 60 if @reconnection_delay > 60
           return
         end
 
+        @reconnection_delay = 1
+
         data = JSON.parse(response.body, symbolize_names: true)
 
         @invocation_id = 0 if @invocation_id > 9095
@@ -100,7 +73,7 @@ class W3DHub
 
         logger.debug(LOG_TAG) { "Connecting to websocket..." }
         this = self
-        @ws = WebSocket::Client::Simple.connect(endpoint, headers: Api::DEFAULT_HEADERS) do |ws|
+        @ws = WebSocketClient.new.connect(endpoint, headers: Api::DEFAULT_HEADERS) do |ws|
           ws.on(:open) do
             logger.debug(LOG_TAG) { "Requesting json protocol, v1..." }
             ws.send({ protocol: "json", version: 1 }.to_json + "\x1e")
@@ -115,65 +88,71 @@ class W3DHub
           end
 
           ws.on(:message) do |msg|
-            msg = msg.data.split("\x1e").first
+            msg = msg.to_str.split("\x1e").first
 
             hash = JSON.parse(msg, symbolize_names: true)
 
             # pp hash if hash[:target] != "ServerStatusChanged" && hash[:type] != 6 && hash[:type] != 3
 
             # Send PING(?)
-            if hash.empty? || hash[:type] == 6
-              ws.send({ type: 6 }.to_json + "\x1e")
-            else
-              case hash[:type]
-              when 1
-                case hash[:target]
-                when "ServerRegistered"
-                  data = hash[:arguments].first
+            if hash.empty? || hash[:type] == TYPE_PING
+              ws.send({ type: TYPE_PING }.to_json + "\x1e")
+              next
+            end
 
-                  this.invocation_id += 1
-                  out = { "type": 1, "invocationId": "#{this.invocation_id}", "target": "SubscribeToServerStatusUpdates", "arguments": [data[:id], 1] }
-                  ws.send(out.to_json + "\x1e")
+            case hash[:type]
+            when 1
+              case hash[:target]
+              when "ServerRegistered"
+                data = hash[:arguments].first
 
-                  BackgroundWorker.foreground_job(
-                    ->(data) { [Api.server_details(data[:id], 2), data] },
-                    ->(array) do
-                      server_data, data = array
+                this.invocation_id += 1
+                out = {
+                        "type": 1,
+                        "invocationId": "#{this.invocation_id}",
+                        "target": "SubscribeToServerStatusUpdates",
+                        "arguments": [data[:id], 1]
+                      }
+                ws.send(out.to_json + "\x1e")
 
-                      next unless server_data
+                BackgroundWorker.foreground_job(
+                  ->(data) { [Api.server_details(data[:id], 2), data] },
+                  ->(array) do
+                    server_data, data = array
 
-                      data[:status] = server_data
+                    next unless server_data
 
-                      server = ServerListServer.new(data)
-                      Store.server_list.push(server)
-                      States::Interface.instance&.update_server_browser(server, :update)
-                    end,
-                    nil,
-                    data
-                  )
+                    data[:status] = server_data
 
-                when "ServerStatusChanged"
-                  id, data = hash[:arguments]
-                  server = Store.server_list.find { |s| s.id == id }
-                  server_updated = server&.update(data)
+                    server = ServerListServer.new(data)
+                    Store.server_list.push(server)
+                    States::Interface.instance&.update_server_browser(server, :update)
+                  end,
+                  nil,
+                  data
+                )
 
-                  BackgroundWorker.foreground_job(->(server) { server }, ->(server) { States::Interface.instance&.update_server_browser(server, :update) }, nil, server) if server_updated
+              when "ServerStatusChanged"
+                id, data = hash[:arguments]
+                server = Store.server_list.find { |s| s.id == id }
+                server_updated = server&.update(data)
 
-                when "ServerUnregistered"
-                  id = hash[:arguments].first
-                  server = Store.server_list.find { |s| s.id == id }
+                BackgroundWorker.foreground_job(->(server) { server }, ->(server) { States::Interface.instance&.update_server_browser(server, :update) }, nil, server) if server_updated
 
-                  if server
-                    Store.server_list.delete(server)
-                    BackgroundWorker.foreground_job(->(server) { server }, ->(server) { States::Interface.instance&.update_server_browser(server, :remove) }, nil, server)
-                  end
+              when "ServerUnregistered"
+                id = hash[:arguments].first
+                server = Store.server_list.find { |s| s.id == id }
+
+                if server
+                  Store.server_list.delete(server)
+                  BackgroundWorker.foreground_job(->(server) { server }, ->(server) { States::Interface.instance&.update_server_browser(server, :remove) }, nil, server)
                 end
               end
             end
           end
 
-          ws.on(:close) do |e|
-            logger.error(LOG_TAG) { e }
+          ws.on(:close) do
+            logger.error(LOG_TAG) { "Connection closed." }
             this.auto_reconnect = true
             ws.close
           end
@@ -218,14 +197,24 @@ class W3DHub
           # unsubscribe from removed servers
           removed_servers.each do
             @invocation_id += 1
-            out = { "type": 1, "invocationId": "#{@invocation_id}", "target": "SubscribeToServerStatusUpdates", "arguments": [server.id, 0] }
+            out = {
+                    "type": 1,
+                    "invocationId": "#{@invocation_id}",
+                    "target": "SubscribeToServerStatusUpdates",
+                    "arguments": [server.id, 0]
+                  }
             ws.send(out.to_json + "\x1e")
           end
 
           # subscribe to new servers
           new_servers.each do
             @invocation_id += 1
-            out = { "type": 1, "invocationId": "#{@invocation_id}", "target": "SubscribeToServerStatusUpdates", "arguments": [server.id, 1] }
+            out = {
+                    "type": 1,
+                    "invocationId": "#{@invocation_id}",
+                    "target": "SubscribeToServerStatusUpdates",
+                    "arguments": [server.id, 1]
+                  }
             ws.send(out.to_json + "\x1e")
           end
         end

lib/websocket_client.rb

@@ -0,0 +1,68 @@
+class W3DHub
+ class WebSocketClient
+   def initialize
+     @errored = nil
+     @connection = nil
+
+     @events = {
+       open: nil,
+       message: nil,
+       close: nil,
+       error: nil
+     }
+   end
+
+   def connect(endpoint, headers: nil, &block)
+     yield(self)
+
+     Sync do |task|
+      endpoint = Async::HTTP::Endpoint.parse(endpoint, alpn_protocols: Async::HTTP::Protocol::HTTP11.names)
+
+      Async::WebSocket::Client.connect(endpoint, headers: headers) do |connection|
+        @connection = connection
+
+        @events[:open]&.call
+
+        while message = connection.read
+          @events[:message].call(message)
+        end
+      # FIXME: Don't rescue for all ta errors?
+      rescue => error
+        @errored = true
+        @events[:error]&.call(error)
+      ensure
+        @events[:close]&.call unless @errored
+        @connection = nil
+        @errored = false
+      end
+     end
+
+     self
+   end
+
+   def on(event, &block)
+     raise "Event must be a symbol" unless event.is_a?(Symbol)
+     raise "Unknown event: #{event.inspect}" unless @events.keys.include?(event)
+     raise "No block given for #{event.inspect}" unless block_given?
+
+     @events[event] = block
+   end
+
+   def send(data, type: :text)
+     @connection&.write(data)
+     @connection&.flush
+   end
+
+   def close
+     @connection&.close
+   end
+
+   def open?
+     !closed?
+   end
+
+   def closed?
+     @connection&.closed?
+   end
+ end
+end

w3d_hub_linux_launcher.rb

@@ -87,7 +87,6 @@ class W3DHub
   BLACK_IMAGE = Gosu::Image.from_blob(1, 1, "\x00\x00\x00\xff")
 end
 
-require "websocket-client-simple"
 require "English"
 require "sdl2"
 
@@ -110,6 +109,7 @@ require_relative "lib/ico"
 require_relative "lib/broadcast_server"
 require_relative "lib/hardware_survey"
 require_relative "lib/game_settings"
+require_relative "lib/websocket_client"
 require_relative "lib/background_worker"
 require_relative "lib/application_manager"
 require_relative "lib/application_manager/manifest"