Class: Bsdkrun::WsClient

Inherits:
Object
  • Object
show all
Defined in:
lib/bsdkrun/ws_client.rb

Overview

A hand-rolled graphql-transport-ws client (the graphql-ws project's protocol; the negotiated subprotocol string is literally "graphql-transport-ws"), because the Ruby standard library has no WebSocket client.

One socket per instance, opened lazily on the first #subscribe and closed once the last subscription ends. A background reader Thread pumps frames off the socket and dispatches to each subscription's callbacks — this SDK is otherwise synchronous (see Sandbox, which shells out via Open3), so the public surface here stays callback-based and lets callers like Client#exec block on a Queue instead of needing their own event loop.

Two mutexes, deliberately not one: @state_mutex guards the small bits of bookkeeping (+@subs+, @pending, @acked) so the reader thread and callers never race on them, while @write_mutex only serializes actual socket writes. Holding one lock across a blocking socket write while the other is needed just for a hash lookup would let a slow write stall unrelated bookkeeping.

Instance Method Summary collapse

Constructor Details

#initialize(ws_url:, token:) ⇒ WsClient

Returns a new instance of WsClient.

Parameters:

  • ws_url (String) —

    e.g. "ws://host:50052/graphql/ws".

  • token (String) 


39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/bsdkrun/ws_client.rb', line 39

def initialize(ws_url:, token:)
  @ws_url = ws_url
  @token = token

  @connect_mutex = Mutex.new
  @write_mutex = Mutex.new
  @state_mutex = Mutex.new

  @socket = nil
  @reader_thread = nil
  @acked = false
  @subs = {}
  @pending = []
  @next_id = 0
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Close the socket immediately, regardless of open subscriptions. Used by Client teardown paths; normal use tears itself down when the last subscription unsubscribes.



115
116
117
 
# File 'lib/bsdkrun/ws_client.rb', line 115

def close
  @connect_mutex.synchronize { close_socket }
end

#connected? ⇒ Boolean

Returns whether the socket is currently open.

Returns:

  • (Boolean) —

    whether the socket is currently open.



107
108
109
 
# File 'lib/bsdkrun/ws_client.rb', line 107

def connected?
  !@socket.nil?
end

#subscribe(query, variables, on_next:, on_error: nil, on_complete: nil) ⇒ Proc

Start a subscription.

Connects (and sends connection_init) on the first call. If connection_ack has not arrived yet, the subscribe message is queued and flushed once it does — starting a subscription before the ack would be sent to a daemon not yet ready to accept operations.

Parameters:

  • query (String)
  • variables (Hash)
  • on_next (#call) —

    invoked with the data hash for each next.

  • on_error (#call, nil) (defaults to: nil) —

    invoked with an Error exception once.

  • on_complete (#call, nil) (defaults to: nil) —

    invoked with no args once, on a clean end.

Returns:

  • (Proc) —

    call to unsubscribe (sends complete, then closes the socket if this was the last subscription).



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/bsdkrun/ws_client.rb', line 69

def subscribe(query, variables, on_next:, on_error: nil, on_complete: nil)
  ensure_connected

  id = (@state_mutex.synchronize { @next_id += 1 }).to_s
  sub = {
    on_next: on_next,
    on_error: on_error || ->(_e) {},
    on_complete: on_complete || -> {}
  }
  start = lambda do
    send_json({ id: id, type: "subscribe", payload: { query: query, variables: variables } })
  end

  ready = @state_mutex.synchronize do
    @subs[id] = sub
    if @acked
      true
    else
      @pending << start
      false
    end
  end
  start.call if ready

  unsubscribed = false
  lambda do
    next if unsubscribed

    unsubscribed = true
    existed = @state_mutex.synchronize { !!@subs.delete(id) }
    next unless existed

    send_json({ id: id, type: "complete" }) if connected?
    close_if_idle
  end
end