Class: Ask::AppServer::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/ask/app_server/connection.rb

Overview

A client connection to the session host: an input/output pair (stdio or a unix socket) plus the connection's per-session event delivery cursors.

Delivery is cursor-based: the connection tracks the last delivered seq for each subscribed session, so any number of clients can attach to the same sessions and each receives exactly the events after its own cursor. Subscribe with afterSeq to replay; clients dedup by seq.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(input, output) ⇒ Connection

Returns a new instance of Connection.

Parameters:

  • input (IO) —

    source of NDJSON lines (responds to gets)

  • output (IO) —

    sink for NDJSON responses/notifications



21
22
23
24
25
26
 
# File 'lib/ask/app_server/connection.rb', line 21

def initialize(input, output)
  @input = input
  @output = output
  @subscriptions = {}
  @mutex = Mutex.new
end

Instance Attribute Details

#subscriptions ⇒ Object (readonly)

session_id => last delivered seq



17
18
19
 
# File 'lib/ask/app_server/connection.rb', line 17

def subscriptions
  @subscriptions
end

Instance Method Details

#advance(session_id, seq) ⇒ Object

Advance the delivery cursor for a session (only while subscribed).



60
61
62
63
64
 
# File 'lib/ask/app_server/connection.rb', line 60

def advance(session_id, seq)
  @mutex.synchronize do
    @subscriptions[session_id] = seq if @subscriptions.key?(session_id)
  end
end

#close ⇒ Object

Drop all subscriptions. Does not close the underlying I/O — the owner (the server) manages IO lifecycle.



68
69
70
 
# File 'lib/ask/app_server/connection.rb', line 68

def close
  @mutex.synchronize { @subscriptions.clear }
end

#cursor(session_id) ⇒ Object

The last delivered seq for a session (0 = replay from the start).



55
56
57
 
# File 'lib/ask/app_server/connection.rb', line 55

def cursor(session_id)
  @mutex.synchronize { @subscriptions[session_id] || 0 }
end

#read_line ⇒ Object

Read one line from the input, or nil on EOF/disconnect.



29
30
31
 
# File 'lib/ask/app_server/connection.rb', line 29

def read_line
  @input.gets
end

#subscribe(session_id, after_seq: 0) ⇒ Object

Subscribe to a session, starting delivery after after_seq.



45
46
47
 
# File 'lib/ask/app_server/connection.rb', line 45

def subscribe(session_id, after_seq: 0)
  @mutex.synchronize { @subscriptions[session_id] = after_seq.to_i }
end

#subscribed?(session_id) ⇒ Boolean

Whether the connection is subscribed to a session.

Returns:

  • (Boolean) 


40
41
42
 
# File 'lib/ask/app_server/connection.rb', line 40

def subscribed?(session_id)
  @mutex.synchronize { @subscriptions.key?(session_id) }
end

#unsubscribe(session_id) ⇒ Object

Unsubscribe from a session.



50
51
52
 
# File 'lib/ask/app_server/connection.rb', line 50

def unsubscribe(session_id)
  @mutex.synchronize { @subscriptions.delete(session_id) }
end

#write(msg) ⇒ Object

Write one JSON-RPC message (Hash) to the output.



34
35
36
37
 
# File 'lib/ask/app_server/connection.rb', line 34

def write(msg)
  @output.puts(JSON.generate(msg))
  @output.flush
end