Initial support for GSH websocket live server updates, minor refactoring to remove thread usage from Worker 'server-side'

lib/common.rb

@@ -0,0 +1,9 @@
+module W3DHubLauncher
+  def self.ca_bundle_path
+    [
+      "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu
+      "/etc/pki/tls/certs/ca-bundle.crt",   # RHEL/Fedora/CentOS
+      "/etc/ssl/ca-bundle.pem"              # Some other distros
+    ].find { |path| File.exist?(path) }
+  end
+end

lib/websocket_client.rb

@@ -0,0 +1,76 @@
+module W3DHubLauncher
+  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|
+        ssl_context = W3DHubLauncher.ca_bundle_path ? OpenSSL::SSL::SSLContext.new : nil
+        # ssl_context&.alpn_protocols = Async::HTTP::Protocol::HTTP11.names
+        ssl_context&.set_params(
+          ca_file: W3DHubLauncher.ca_bundle_path,
+          verify_mode: OpenSSL::SSL::VERIFY_PEER
+        )
+
+        # endpoint = Async::HTTP::Endpoint.parse(endpoint, alpn_protocols: Async::HTTP::Protocol::HTTP11.names, ssl_context: ssl_context)
+        endpoint = Async::HTTP::Endpoint.parse(endpoint, ssl_context: ssl_context)
+
+        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

lib/worker.rb

@@ -18,6 +18,9 @@ module W3DHubLauncher
 
       @buffer = StringIO.new
       @socket = UNIXSocket.new(IPC_PATH)
+    rescue Errno::ENOENT
+      sleep 1
+      retry
     end
 
     def connected?
@@ -29,7 +32,6 @@ module W3DHubLauncher
     end
 
     def init_server
-      @threads = []
       @requests = []
       @tasks = []
 
@@ -39,34 +41,17 @@ module W3DHubLauncher
       @game_servers = []
       @host_pings = {}
 
-      # connect to and monitor GSH web service
-      @threads << Thread.new { game_server_hub_websocket }
-      # connect to and monitor Backend web service
-      @threads << Thread.new { backend_websocket }
-      # poke game servers to ascertain their ping relative to the local machine
-      @threads << Thread.new { ping_game_servers }
-
       @w3dhub_api = W3DHubLauncher::W3DHubApi.new
 
       Async do |task|
-        task.async do
-          while true
-            @tasks.delete_if { |t| [Request::STATUS_ERROR, Request::STATUS_COMPLETE].include?(t.status) }
-
-            if @tasks.first && @tasks.all? { |t| t.status == Request::STATUS_PENDING }
-              @tasks.first.start(self)
-            end
-
-            sleep 0.25
-          end
-        end
-
         task.async do |subtask|
           UNIXServer.open(IPC_PATH) do |server|
+            puts "IPC running!"
+
             while(socket = server.accept)
               @client = socket
 
-              task.async do
+              subtask.async do
                 while(data = socket.gets)
                   json = JSON.parse(data)
                   query = Request::Query.new(type: json["type"].to_sym, request_id: json["request_id"], data: json["data"])
@@ -88,10 +73,96 @@ module W3DHubLauncher
         ensure # manually delete "socket"
           File.delete(IPC_PATH)
         end
+
+        # Handle getting and updating game server data
+        task.async do
+          game_server_hub_websocket
+        end
+
+        # Handle backend communication
+        task.async do
+          backend_websocket
+        end
+
+        # Handle getting ping times to servers
+        task.async do
+          ping_game_servers
+        end
+
+        # Handle Task requests
+        task.async do
+          while true
+            @tasks.delete_if { |t| [Request::STATUS_ERROR, Request::STATUS_COMPLETE].include?(t.status) }
+
+            if @tasks.first && @tasks.all? { |t| t.status == Request::STATUS_PENDING }
+              @tasks.first.start(self)
+            end
+
+            sleep 0.25
+          end
+        end
       end
     end
 
     def game_server_hub_websocket
+      WebSocketClient.new.connect("ws://localhost:9292/api/websocket") do |websocket|
+        websocket.on(:open) do
+          # puts "CONNECTED"
+
+          websocket.send( { "protocol_version": 1 }.to_json )
+        end
+
+        websocket.on(:message) do |message|
+          # puts "MESSAGE"
+          # pp message
+
+          hash = JSON.parse(message)
+
+          case hash["type"]
+          when "ping"
+            websocket.send({ "type":"pong", "data":"" }.to_json)
+          when "pong"
+          when "server_list"
+            @game_servers = hash["data"].map { |server| Worker::Api::GameServer.new(server) }
+            broadcast(:game_server_list, data: hash["data"])
+
+          when "server_added", "server_updated"
+            existing_game_server = @game_servers.find { |s| s.id == hash["data"]["id"] }
+            game_server = Worker::Api::GameServer.new(hash["data"])
+
+
+            if existing_game_server # replace
+              @game_servers[@game_servers.index(existing_game_server)] = game_server
+            else # add
+              @game_servers << game_server
+            end
+            broadcast(hash["type"].to_sym, data: hash["data"])
+
+          when "server_removed"
+            game_server = @game_servers.find { |s| s.id == hash["data"]["id"] }
+            @game_servers.delete(game_server) if game_server
+            broadcast(hash["type"].to_sym, data: hash["data"]["id"])
+
+          when "server_details"
+            # TODO
+          end
+        end
+
+        websocket.on(:close) do
+          # puts "CLOSED"
+          sleep 5
+          game_server_hub_websocket
+        end
+
+        websocket.on(:error) do |error|
+          puts "ERROR: #{error}"
+
+          puts error.backtrace
+
+          sleep 5
+          game_server_hub_websocket
+        end
+      end
     end
 
     def backend_websocket
@@ -101,8 +172,8 @@ module W3DHubLauncher
       socket = Socket.open(Socket::AF_INET, Socket::SOCK_DGRAM, Socket::IPPROTO_ICMP)
       socket.setsockopt(:SOCKET, :TIMESTAMP, true)
 
-      Async do |task|
-        while true
+      Sync do |task|
+        loop do
           echo_requests = {}
 
           # Send pings to each unique server host

w3d_hub_linux_launcher.rb

@@ -23,6 +23,7 @@ require "chunky_png"
 require "sdl3"
 
 require_relative "lib/gosu_ext"
+require_relative "lib/common"
 
 require_relative "lib/version"
 require_relative "lib/constants"
@@ -49,6 +50,7 @@ require_relative "lib/application_helper"
 require_relative "lib/ico"
 require_relative "lib/dds"
 require_relative "lib/ww_mix"
+require_relative "lib/websocket_client"
 
 require_relative "lib/worker"
 require_relative "lib/worker/api"
@@ -78,9 +80,6 @@ module W3DHubLauncher
     W3DHubLauncher::Worker.new.listen
   end
 
-  # Prevent race condition where server IPC isn't ready before the client attempts to connect
-  sleep 0.1
-
   # UNIXSocket / client
   WORKER = W3DHubLauncher::Worker.new
   WORKER.connect