Julia - how to subscribe via WebSockets

Viewed 414

I would like to subscribe to some data feed using Websockets using Julia.

For example, from the linux terminal, I can successfully get data like this:

wscat -c wss://www.bitmex.com/realtime
{"op": "subscribe", "args": ["orderBookL2_25:XBTUSD"]}

Now in Julia, I cannot find a solution. I have tried the following, but it crashes Julia:

using WebSockets, JSON

uri = "wss://www.bitmex.com/realtime"
json_part = "{'op': 'subscribe', 'args': ['orderBookL2_25:XBTUSD']}"

inbox = Channel{String}(10)
outbox = Channel{String}(10)

ws_task = @async WebSockets.open(uri) do ws
  while isopen(ws)
        inbox_task = @async while !eof(ws)
            put!(inbox, String(read(ws)))
        end
        outbox_task = @async while isopen(ws)
            write(ws, take!(outbox))
        end
    end
end

# here Julia is crashing (hangs forever, I cannot get the cursor back)

put!(outbox, json_part)
take!(inbox)

Can someone help to get a working solution to subscribe to data feed using Julia?

2 Answers

Just remove the outer while loop and you're good to go:

julia> using WebSockets, JSON

julia> uri = "wss://www.bitmex.com/realtime"
"wss://www.bitmex.com/realtime"

julia> json_part = "{'op': 'subscribe', 'args': ['orderBookL2_25:XBTUSD']}"
"{'op': 'subscribe', 'args': ['orderBookL2_25:XBTUSD']}"

julia> inbox = Channel{String}(10)
Channel{String}(sz_max:10,sz_curr:0)

julia> outbox = Channel{String}(10)
Channel{String}(sz_max:10,sz_curr:0)

julia> ws_task = @async WebSockets.open(uri) do ws
           inbox_task = @async while !eof(ws)
               put!(inbox, String(read(ws)))
           end
           outbox_task = @async while isopen(ws)
               write(ws, take!(outbox))
           end
       end
Task (runnable) @0x00000000135b3990

julia> put!(outbox, json_part)
"{'op': 'subscribe', 'args': ['orderBookL2_25:XBTUSD']}"

julia> take!(inbox)
"{\"info\":\"Welcome to the BitMEX Realtime API.\",\"version\":\"2020-10-06T22:31:35.000Z\",\"timestamp\":\"2020-10-26T16:56:02.455Z\",\"docs\":\"https://www.bitmex.com/app/wsAPI\",\"limit\":{\"remaining\":38}}"

With the outer while isopen(ws) loop in place you're continously creating new inbox/outbox_tasks. I suspect you wanted to restart them if the WS connection drops or something, but you'll need to handle that differently.

First solution with the subscribe inside the url (not always possible or desirable):

using WebSockets, JSON

uri = "wss://www.bitmex.com/realtime?subscribe=trade:XBT"

function open_websocket() 
  WebSockets.open(uri) do ws
    while isopen(ws)
      data, success = readguarded(ws)
      if success
        data = JSON.parse(String(data))
        print(data, "\n")
      end
    end

    if !isopen(ws)
      @async open_websocket()
    end

  end
end

@async open_websocket()

Second solution with subscribe after the socket is opened:

using WebSockets, JSON

uri = "wss://www.bitmex.com/realtime"

payload = Dict(
               :op => "subscribe",
               :args => "trade:XBT"
           )

function open_websocket() 
   WebSockets.open(uri) do ws
     if isopen(ws)
       write(ws, JSON.json(payload))
     end

     while isopen(ws)
       data, success = readguarded(ws)
       if success
         data = JSON.parse(String(data))
         print(data, "\n")
       end
     end

     if !isopen(ws)
       @async open_websocket()
     end

   end
 end
       
@async open_websocket()
Related