Class: Terminalwire::V2::Mux

Inherits:
Object
  • Object
show all
Defined in:
lib/terminalwire/v2/mux.rb

Overview

Allocates stream ids and correlates in-flight requests to their responses. The starting id is injectable so recorded vectors replay deterministically.

Instance Method Summary collapse

Constructor Details

#initialize(start: 1) ⇒ Mux

Returns a new instance of Mux.

Raises:

  • (ArgumentError) 


7
8
9
10
11
12
13
14
15
 
# File 'lib/terminalwire/v2/mux.rb', line 7

def initialize(start: 1)
  raise ArgumentError, "start must be >= 1 (0 is the control stream)" if start < 1

  @next = start
  @pending = {}
  # The runtime allocates/registers from the caller thread while the read
  # pump resolves from its own thread, so the registry is mutex-guarded.
  @mutex = Mutex.new
end

Instance Method Details

#allocate ⇒ Object

Allocate a fresh stream id.



18
19
20
21
22
23
24
 
# File 'lib/terminalwire/v2/mux.rb', line 18

def allocate
  @mutex.synchronize do
    sid = @next
    @next += 1
    sid
  end
end

#pending?(sid) ⇒ Boolean

Returns:

  • (Boolean) 


31
32
33
 
# File 'lib/terminalwire/v2/mux.rb', line 31

def pending?(sid)
  @mutex.synchronize { @pending.key?(sid) }
end

#pending_count ⇒ Object 



44
45
46
 
# File 'lib/terminalwire/v2/mux.rb', line 44

def pending_count
  @mutex.synchronize { @pending.size }
end

#register(sid, context = nil) ⇒ Object

Mark a request stream as awaiting a response, stashing caller context.



27
28
29
 
# File 'lib/terminalwire/v2/mux.rb', line 27

def register(sid, context = nil)
  @mutex.synchronize { @pending[sid] = context }
end

#resolve(sid) ⇒ Object

Resolve a pending request, returning (and removing) its context.



36
37
38
39
40
41
42
 
# File 'lib/terminalwire/v2/mux.rb', line 36

def resolve(sid)
  @mutex.synchronize do
    raise ProtocolError, "response for unknown stream #{sid}" unless @pending.key?(sid)

    @pending.delete(sid)
  end
end