Class: HermesAgent::Client::Stream

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/hermes_agent/client/stream.rb

Overview

A consumable Server-Sent Events stream.

Stream parses the SSE frames emitted by a streaming endpoint, wrapping each frame's data payload (parsed as JSON) in an event wrapper object, and implements the block-or-enumerator contract:

  • Iterated with a block (via #each), it yields each event as it arrives, giving natural backpressure over the network read.
  • Without a block it is an Enumerable the caller drives itself.

After the stream is fully consumed, #result returns the aggregated final object (built by the aggregator block given at construction). It is a single-pass stream over a live network read: it can be iterated once.

The class is HTTP-agnostic — it consumes anything that yields String byte chunks via #each (the http gem's response body, or an array of chunks in tests) — so it never owns or manages the network connection itself.

Instance Method Summary collapse

Instance Method Details

#each {|event| ... } ⇒ self, Enumerator

Iterate the events. With a block, yields each event as it is parsed and returns self. Without a block, returns an Enumerator.

Yield Parameters:

  • event (Entity) —

    Each parsed event, in order.

Returns:

  • (self, Enumerator)

Raises:

  • (Error) —

    If the stream has already been consumed.



75
76
77
78
79
80
81
82
 
# File 'lib/hermes_agent/client/stream.rb', line 75

def each(&block)
  return enum_for(:each) unless block

  raise Error, "Stream has already been consumed" if @consumed

  consume(&block)
  self
end

#result ⇒ Object?

The aggregated final object, available after the stream is consumed. Consuming the stream first if it has not been iterated.

Returns:

  • (Object, nil) —

    Whatever the aggregator block returned, or nil when no aggregator was given.



91
92
93
94
 
# File 'lib/hermes_agent/client/stream.rb', line 91

def result
  consume unless @consumed
  @result
end