Class: Cadenya::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/cadenya/sse.rb

Constant Summary collapse

MAX_RECONNECTS =

Bounded: 5 consecutive attempts per outage, 500ms*2^n capped 10s (the server's retry: hint overrides). Sliced sleeps keep close() prompt.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(decoder, last_event_id: nil, skip_events: [], auto_reconnect: true, &perform) ⇒ Stream

perform runs the HTTP request: it receives a Core::CancelHandle to register transport teardown on, and calls its block with each chunk.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/cadenya/sse.rb', line 26

def initialize(decoder, last_event_id: nil, skip_events: [], auto_reconnect: true, &perform)
  @decoder = decoder
  @perform = perform
  @last_event_id = last_event_id
  # Housekeeping event names (the `event:` field) skipped without
  # decoding; their `id:` fields still advance the resume checkpoint.
  @skip_events = skip_events.to_a.freeze
  # Auto-reconnect (EventSource semantics): a MID-STREAM transport drop
  # re-runs @perform from the resume checkpoint. Clean EOF, close, and
  # budget exhaustion never reconnect.
  @auto_reconnect = auto_reconnect
  @retry_hint_ms = nil
  @reconnect_attempts = 0
  @closed = false
  @consumed = false
  @lifecycle = Mutex.new
  @cancel = Core::CancelHandle.new
end

Instance Attribute Details

#last_event_id ⇒ Object (readonly)

The resume checkpoint: seeded from the id this stream was resumed with, then updated by id: fields (persistent per the SSE spec). Pass it as last_event_id: to resume after a disconnect.



22
23
24
 
# File 'lib/cadenya/sse.rb', line 22

def last_event_id
  @last_event_id
end

Instance Method Details

#close ⇒ Object

Stop the stream deterministically — safe to call from the consuming block or another thread, INCLUDING while the consumer is blocked on a silent socket: the cancel handle tears the transport down, which unblocks the read immediately. Idempotent; deliberate close is normal termination, never a transport error.



50
51
52
53
54
55
56
57
 
# File 'lib/cadenya/sse.rb', line 50

def close
  handle = @lifecycle.synchronize do
    @closed = true
    @cancel
  end
  handle.cancel!
  nil
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


59
60
61
 
# File 'lib/cadenya/sse.rb', line 59

def closed?
  @closed
end

#each ⇒ Object 



93
94
95
96
97
98
99
 
# File 'lib/cadenya/sse.rb', line 93

def each
  return enum_for(:each) unless block_given?

  each_event { |event| yield event.data }
  # Ruby collection iterators return the receiver in block form.
  self
end

#each_event ⇒ Object 



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/cadenya/sse.rb', line 101

def each_event
  return enum_for(:each_event) unless block_given?

  # A Stream wraps ONE HTTP request. Re-enumerating would silently reopen
  # the connection with the ORIGINAL resume header (stale checkpoint) and
  # duplicate events; reconnection is explicit — construct a new stream
  # with `last_event_id: old.last_event_id`. A stream closed before
  # iteration yields nothing and never opens a connection.
  @lifecycle.synchronize do
    return if @closed && !@consumed
    raise IOError, "stream already consumed — reconnect with a new stream using last_event_id: #{@last_event_id.inspect}" if @consumed

    @consumed = true
  end

  buffer = +""
  data_lines = []
  event_name = nil

  flush = proc do
    # Authoritative close check BEFORE decoding: several complete events
    # can arrive in one chunk, and no buffered event may be decoded or
    # yielded once close (from the consumer block or elsewhere) returns.
    raise Closed if @closed

    unless data_lines.empty? || @skip_events.include?(event_name)
      payload = JSON.parse(data_lines.join("\n"))
      # Per the SSE spec the last-event-ID buffer persists across events
      # until another `id:` field changes it (an empty one resets it).
      yield ServerSentEvent.new(event_name, @decoder.call(payload), @last_event_id)
    end
    data_lines = []
    event_name = nil
    raise Closed if @closed
  end

  handle_line = proc do |line|
    if line.empty?
      flush.call
    elsif !line.start_with?(":")
      field, _, value = line.partition(":")
      value = value.delete_prefix(" ")
      case field
      when "data" then data_lines << value
      when "event" then event_name = value
      when "id"
        # Ids containing U+0000 are ignored per the event-stream
        # algorithm; an empty id resets the buffer.
        @last_event_id = value.empty? ? nil : value unless value.include?("\0")
      when "retry"
        # Reconnection-delay hint, honored during auto-reconnect.
        @retry_hint_ms = [value.to_i, 60_000].min if /\A[0-9]+\z/.match?(value)
      end
    end
  end

  # WHATWG event streams terminate lines with LF, CRLF, OR bare CR; a CR
  # at a chunk boundary must wait for the next chunk to see whether an
  # LF follows (CRLF is one terminator, never two).
  next_line = proc do |at_eof|
    index = buffer.index(/[\r\n]/)
    if index.nil?
      nil
    elsif buffer[index] == "\n"
      line = buffer[0...index]
      buffer.slice!(0..index)
      line
    elsif index == buffer.length - 1 && !at_eof
      nil # possible CRLF split across chunks
    else
      line = buffer[0...index]
      consume = buffer[index + 1] == "\n" ? index + 1 : index
      buffer.slice!(0..consume)
      line
    end
  end

  loop do
    begin
      @perform.call(@cancel, @last_event_id) do |chunk|
        # Bytes flowing again: the reconnect budget is per-outage.
        @reconnect_attempts = 0
        raise Closed if @closed
        buffer << chunk
        while (line = next_line.call(false))
          handle_line.call(line)
        end
      end
    rescue Closed
      return
    rescue APIConnectionError
      # Only a MID-STREAM transport drop (or a failed reconnect
      # handshake, which raises the same family and consumes budget)
      # reconnects; HTTP-level failures propagate untouched.
      raise unless reconnect_backoff

      buffer = +""
      data_lines = []
      event_name = nil
      next
    end
    break # clean EOF: API streams may legitimately end
  end
  # A deliberate close may surface as a NORMAL transport return (the
  # cancel handle maps teardown to clean termination); the leftover
  # buffer must still never be processed.
  return if @closed

  while (line = next_line.call(true))
    handle_line.call(line)
  end
  handle_line.call(buffer) unless buffer.empty?
  flush.call
rescue Closed
  nil
end