Async WebSockets

[oliviermilla]

• Julia 1.8.2
• HTTP.jl 1.5.1
• MbedTLS.jl 1.1.7

Hello everyone,

I am building a package to connect to a websocket service.

Here is my understanding of the WebSockets API, the handler function is what the end user would pass to get callbacks.

function open(handler::Function)
    WebSockets.open(WEBSOCKETS_URL) do ws
        global webSocket[] = ws

        while true
            response = receive(ws)
            handler(JSON3.read(response, Response))
        end
    end
end

function close()
    WebSockets.close(webSocket[])
end

the while true feels really bad I have to say. Isn't there an async API to provide HTTP.WebSockets with callbacks directly?

Love your feedback, just starting out in julia.
Best!

[quinnj]

I'd recommend taking a look at the WebSockets documentation; particularly, we recommend this following pattern:

WebSockets.open(url) do ws
    for msg in ws
        # do cool stuff with msg
    end
end

For async, you can just wrap that block with the @async macro and store the task to inspect later if needed:

websocket_client_task = @async WebSockets.open(url) do ws
    for msg in ws
        # do cool stuff with msg
    end
end

[oliviermilla]

Hello @quinnj thank you for the reply, on my way to discovering tasks and channels then. :)

[oliviermilla]

After much learning, I still find the current API to badly embedd with Tasks when desiring bi-directional commnication.

function receive(c::Channel)
    WebSockets.open(url) do ws
    # TODO send subscriptions
    for msg in ws
        # do cool stuff with msg
    end
end


channel = Channel(receive)

In order to pass the subscription parameters (the TODO), I don't see how without using a module global variable (which sucks).
My other idea is to have a task build using 2 channels, for which I'll have to bind and manage the life cycle. One channel to get subscription messages in, one to get the streams.

I'd value a feedback on the approach. Maybe I'm missing something.

In any case, ❤️

[quinnj]

Sorry for the slow response. I'm not sure I quite follow your example there; if you can share a bit more context about the larger workflow you're trying to achieve, I can probably help suggest something more concrete.

[oliviermilla]

Hello @quinnj, my turn to have (really slow) response.

Here is the broader context:

• I want to receive websocket data asynchronously. My idea was to go for channels for sake of simplicity.
• Subscription parameters to the WebSocket are programmable. Meaning I need to pass the initial message for the subscription.
• Ideally being able to send subscribe/unsubscribe messages during the lifetime of the socket.

That led me to the design posted in my previous post where I define a receive function that is used by a channel. But then, how I pass subscription messages into the channel is where I'm stuck. :)

[likev]

I don't think there is an async API to provide HTTP.WebSockets with callbacks.
the while true is what for msg in ws do internal, maybe you can define a function and returns a closure

function create_ws(WEBSOCKETS_URL, subscriptions, callback)
    return function()
        WebSockets.open(WEBSOCKETS_URL) do ws
            global webSocket[] = ws
            # TODO send subscriptions
            while true
                response = receive(ws)
                callback(JSON3.read(response, Response))
                sleep(0.05) # sleep a little
            end
        end
    end
end