Best way to subscribe/manage events and messages

[j-a-m-l]

I've seen that this project was using websocket-driver but it was replaced by a new implementation. I've not found any reason for that, so would you mind to explain the rationale behind it? I'd like to understand the issues that it involves.

Related to that, what is the recommended way to manage the lifecycle of a client connection? I'd like to wrap the client in way that my library would process the messages, but also it would allow other developers to subscribe or intercept the events (connected, disconnected) and messages.

I mean, using websocket-driver, it would be something like:

driver.on :open do |_event|
  emit :connected
end
driver.on :close do |_event|
  emit :disconnected
end
driver.on :error do |event|
  emit :error, event
end
driver.on :message do |event|
  emit :message, event.data
end

This was very handy to capture errors, log when the connection is closed and try to reconnect, etc.

But, when using something like this simple example:

endpoint = Async::HTTP::Endpoint.parse url

Async do |task|
  # Connection has not been established yet

  Async::WebSocket::Client.connect endpoint do |connection|
    # Since connection is ready: could this be used to emit a `connected` event?
    emit :connected

    while message = connection.read
      # Message has been received
      emit :message, message
    end
  end
end

I don't know how it's supposed that developers react to errors, disconnections or other events.

[ioquatix]

websocket-driver required raw I/O in order to function, which obviously won't work with HTTP/2. It's event based callbacks leave a lot to be desired in terms of life cycle management. In addition, it was not memory efficient which was one of the goals of the rewrite, which allowed for 1 million simultaneous connections.

One way to deal with your code is to define an interface for client code.

class Plugin
    def connected(connection)
    end

    def disconnected(connection)
    end

    def message(connection, message)
    end

    def closed(connection, error = nil)
    end
end

plugin = Plugin.new
endpoint = Async::HTTP::Endpoint.parse(url)

Async do |task|
    Async::WebSocket::Client.connect(endpoint) do |connection|
        # Since connection is ready: could this be used to emit a `connected` event?
        plugin.connected(connection)

        while message = connection.read
            plugin.message(connection, message)
        end
    ensure
        plugin.closed(connection, $!)
    end
end

However, I think this is a questionable design, because I'm not convinced there is value in the "plugin" style architecture. When I worked on slack-client gem, I found the flow control of the event driven architecture was almost impossible to follow and lead to some ridiculously complicated/unreproducible bugs.

The value of async is it allows your sequential code to be sequential. So my advice is to avoid the layer of indirection that "events" bring and keep your code simple and tightly coupled.

Here is the default implementation of Connection which is instantiated once per connection: https://github.com/socketry/async-websocket/blob/master/lib/async/websocket/connection.rb

You can change the connection by using the handler: option. https://github.com/socketry/async-websocket/blob/master/lib/async/websocket/client.rb#L89

This is per connection. If you want to handle reconnections, you should handle it at that layer and one layer above, i.e. put your websocket connection code in a loop, and if the connection terminates with an error which allows for reconnection, e.g.

class Client
    def connect
        while @running
            endpoint = Async::HTTP::Endpoint.parse(@url)

            Async do |task|
                Async::WebSocket::Client.connect(endpoint) do |connection|
                    # Since connection is ready: could this be used to emit a `connected` event?
                    self.connected(connection)

                    while message = connection.read
                        self.message(connection, message)
                    end
                ensure
                    self.closed(connection, $!)
                end
            end
        end
    end
end

[j-a-m-l]

Thank you for your detailed reply.

I though about a similar event driven implementation, but I'm also aware of the problems that it could provoke, so I may follow your suggestion: would you be interested on another example that covers some of the features that I want to implement in my real project?

[ioquatix]

Sure

[ioquatix]

How did it go?

[j-a-m-l]

I've just started it and now I'm creating something more involved as a way to test some concepts. Probably I'll have time this weekend to finish it.

[j-a-m-l]

I didn't forget this issue!

I created a simple drop-in replacement for the previous event driven implementation. It was enough for being used internally, but it wasn't production ready. Since the project that originally needed WebSocket is deprecated now, I don't see any reason to release that gem publicly, but if someone is truly interested on it, I could paste some of its code here.