Refactor remote connections (#76)

* Refactor remote connections to use the conneciton identifier to save memory. Fixes #75

* Refactor how remote connections are handled. Fixes #75

Author: jwoertink

spec/cable/server_spec.cr

@@ -0,0 +1,22 @@
+require "../spec_helper"
+
+include RequestHelpers
+
+describe Cable::Server do
+  describe "#remote_connections" do
+    it "finds the connection and disconnects it" do
+      Cable.reset_server
+      Cable.temp_config(backend_class: Cable::DevBackend) do
+        socket = DummySocket.new(IO::Memory.new)
+        request = builds_request("abc123")
+        connection = ApplicationCable::Connection.new(request, socket)
+        Cable.server.add_connection(connection)
+        connection.connection_identifier.should contain("abc123")
+
+        Cable.server.remote_connections.find(connection.connection_identifier).disconnect
+        Cable::DevBackend.published_messages.should contain({"cable_internal/#{connection.connection_identifier}", "disconnect"})
+        connection.close
+      end
+    end
+  end
+end

src/cable/connection.cr

@@ -40,6 +40,7 @@ module Cable
         subscribe_to_internal_channel
       rescue e : UnathorizedConnectionException
         reject_connection!
+        unsubscribe_from_internal_channel
         socket.close(HTTP::WebSocket::CloseCode::NormalClosure, "Farewell")
         Cable::Logger.info { ("An unauthorized connection attempt was rejected") }
       end
@@ -168,14 +169,14 @@ module Cable
       server = Cable.server
       channel = self
       spawn(name: "Cable::Connection - subscribe_to_internal_channel") do
-        server.add_internal_subscription(internal_channel, channel)
-        server.backend.open_subscribe_connection(internal_channel)
+        if !channel.connection_rejected?
+          server.backend.open_subscribe_connection(internal_channel)
+        end
       end
     end
 
     private def unsubscribe_from_internal_channel
       Cable.server.backend.unsubscribe(internal_channel)
-      Cable.server.remove_internal_subscription(internal_channel)
     end
   end
 end

src/cable/remote_connections.cr

@@ -11,19 +11,22 @@ module Cable
     #
     # find("1234")
     # ```
-    def find(identifier : String)
+    # NOTE: This code may run on a different machine than where the `@server.connections`
+    # is actually sitting in memory. For this reason, we just pass the value right through
+    # the backend (i.e. redis), and let that broadcast out to all running instances.
+    def find(identifier : String) : RemoteConnection
       RemoteConnection.new(@server, identifier)
     end
 
     private class RemoteConnection
       def initialize(@server : Cable::Server, @value : String)
       end
 
-      def disconnect
+      def disconnect : Nil
         @server.backend.publish_message(internal_channel, Cable.message(:disconnect))
       end
 
-      private def internal_channel
+      private def internal_channel : String
         "cable_internal/#{@value}"
       end
     end

src/cable/server.cr

@@ -23,8 +23,9 @@ module Cable
   class Server
     include Debug
 
+    # The String key is the `connection_identifier` value for `Cable::Connection`
+    getter connections = {} of String => Cable::Connection
     getter errors = 0
-    getter connections = {} of String => Connection
     getter fiber_channel = ::Channel({String, String}).new
     getter pinger : Cable::RedisPinger do
       Cable::RedisPinger.new(self)
@@ -40,7 +41,6 @@ module Cable
     end
 
     @channels = {} of String => Channels
-    @_internal_subscriptions = {} of String => Cable::Connection
     @channel_mutex = Mutex.new
 
     def initialize
@@ -65,18 +65,6 @@ module Cable
       connections.delete(connection_id).try(&.close)
     end
 
-    def add_internal_subscription(internal_channel : String, connection : Cable::Connection)
-      if internal_channel.presence && !connection.closed?
-        @_internal_subscriptions[internal_channel] = connection
-      end
-    end
-
-    def remove_internal_subscription(internal_channel : String)
-      if @_internal_subscriptions.has_key?(internal_channel)
-        @_internal_subscriptions.delete(internal_channel)
-      end
-    end
-
     def subscribe_channel(channel : Channel, identifier : String)
       @channel_mutex.synchronize do
         if !@channels.has_key?(identifier)
@@ -132,12 +120,13 @@ module Cable
       Cable.settings.on_error.call(e, "IO::Error Exception: #{e.message} -> #{self.class.name}#send_to_channels(channel, message)")
     end
 
-    def send_to_internal_channels(channel_identifier : String, message : String)
-      if internal_channel = @_internal_subscriptions[channel_identifier]?
+    def send_to_internal_connections(connection_identifier : String, message : String)
+      if internal_connection = connections[connection_identifier]?
         case message
         when Cable.message(:disconnect)
-          Cable::Logger.info { "Removing connection (#{channel_identifier})" }
-          internal_channel.close
+          Cable::Logger.info { "Removing connection (#{connection_identifier})" }
+          internal_connection.close
+          remove_connection(connection_identifier)
         end
       end
     end
@@ -182,8 +171,10 @@ module Cable
       spawn(name: "Cable::Server - process_subscribed_messages") do
         while received = fiber_channel.receive
           channel, message = received
-          if channel.includes?("cable_internal")
-            server.send_to_internal_channels(channel, message)
+          if channel.starts_with?("cable_internal")
+            identifier = channel.split('/').last
+            connection_identifier = server.connections.keys.find!(&.starts_with?(identifier))
+            server.send_to_internal_connections(connection_identifier, message)
           else
             server.send_to_channels(channel, message)
           end