Class: OKF::MCP::HTTP::Stream

Inherits:
Object
  • Object
show all
Defined in:
lib/okf/mcp/http.rb

Overview

What the SDK writes SSE frames to, adapting its stream contract to WEBrick's proc-body one. The SDK expects write/flush per frame, EPIPE out of write to mean the peer is gone, and close to end the stream; WEBrick ends the response when the body proc returns. So the proc parks in #wait until the SDK — its keepalive thread on a dead peer, or the transport's own close — calls #close, and only then hands the thread back (see #stream_response).

Instance Method Summary collapse

Constructor Details

#initialize(wire) ⇒ Stream

Returns a new instance of Stream.



51
52
53
54
55
56
57
 
# File 'lib/okf/mcp/http.rb', line 51

def initialize(wire)
  @wire = wire
  @state = Mutex.new
  @wire_lock = Mutex.new
  @done = ConditionVariable.new
  @closed = false
end

Instance Method Details

#close ⇒ Object 



83
84
85
86
87
88
 
# File 'lib/okf/mcp/http.rb', line 83

def close
  @state.synchronize do
    @closed = true
    @done.broadcast
  end
end

#flush ⇒ Object

WEBrick's ChunkedWrapper has no flush; each write already reaches the socket as a complete chunk.



78
79
80
81
 
# File 'lib/okf/mcp/http.rb', line 78

def flush
  @wire.flush if @wire.respond_to?(:flush)
  nil
end

#wait ⇒ Object 



90
91
92
 
# File 'lib/okf/mcp/http.rb', line 90

def wait
  @state.synchronize { @done.wait(@state) until @closed }
end

#write(data) ⇒ Object

One frame, one chunk. A dead peer raises EPIPE/ECONNRESET straight out of the socket write — exactly the signal the SDK's stream cleanup keys on, so it must never be swallowed here.

The wire has its own lock, deliberately separate from the state's: a stalled peer (alive, not reading, send buffer full) parks the write with no EPIPE to raise, and #close belongs to shutdown — it must never queue behind a peer's buffer, so it takes only the state lock. A write racing close lands on a socket that is closing anyway; the resulting IOError/EPIPE is the SDK's cleanup signal.



69
70
71
72
73
74
 
# File 'lib/okf/mcp/http.rb', line 69

def write(data)
  @state.synchronize do
    raise IOError, "stream is closed" if @closed
  end
  @wire_lock.synchronize { @wire.write(data) }
end