lib/ksql/stream.rb
def start(&block)
      @headers = {}

      @request.on(:headers) { |headers| @headers = headers }

      @request.on(:body_chunk) do |body|
        next unless body.present?

        response = Ksql::Connection::Response.new(body: body, headers: @headers)
        raise Ksql::StreamError.new(response.body['message']) if response.error?

        if response.body.is_a? Hash
          @event_class = build_event_class(response.body)
          next
        else
          event = @event_class.new(*response.body)
        end

        yield(event)
      end

      @client.call_async(@request)
    end
#
# Streaming connection handler
#
# * Start the stream
# * Wrap the stream events into OpenStruct instances
# * Execute the passed block
#
# @param [Block] &block Code to execute each time an event arrives
#
Streaming connection handler
Start the stream
Wrap the stream events into OpenStruct instances
Execute the passed block
Streaming connection handler Start the stream Wrap the stream events into OpenStruct instances Execute the passed block
def start(&block)
      @headers = {}
      @request.on(:headers) { |headers| @headers = headers }
      @request.on(:body_chunk) do |body|
        next unless body.present?
        response = Ksql::Connection::Response.new(body: body, headers: @headers)
        raise Ksql::StreamError.new(response.body['message']) if response.error?
        if response.body.is_a? Hash
          @event_class = build_event_class(response.body)
          next
        else
          event = @event_class.new(*response.body)
        end
        yield(event)
      end
      @client.call_async(@request)
    end
def start ( & block ) @headers = { } @request . on ( :headers ) { | headers | @headers = headers } @request . on ( :body_chunk ) do | body | next unless body . present? response = Ksql :: Connection :: Response . new ( body : body , headers : @headers ) raise Ksql :: StreamError . new ( response . body [ 'message' ] ) if response . error? if response . body . is_a? Hash @event_class = build_event_class ( response . body ) next else event = @event_class . new ( * response . body ) end yield ( event ) end @client . call_async ( @request ) end
Streaming connection handler
 Start the stream
 Wrap the stream events into OpenStruct instances
 Execute the passed block
Streaming connection handler Start the stream Wrap the stream events into OpenStruct instances Execute the passed block
block
█ Code to execute each time an event arrives
█ Code to execute each time an event arrives