Class: Neo4j::Driver::Bolt::StreamHandler

Inherits:
Object
  • Object
show all
Defined in:
lib/neo4j/driver/bolt/stream_handler.rb

Overview

The wire handler for a streaming PULL/DISCARD: the dedicated reader routes the result's replies here, and it fills the result's RecordBuffer.

Records are forwarded to the buffer incrementally — the instant each is decoded — so the cursor can read record 1 without waiting for the whole batch. The batch's terminating SUCCESS resolves the buffer's has_more promise (batch_complete) or ends the stream (finish); a FAILURE/IGNORED ends it too. The cursor consumes the buffer and drives follow-up PULLs — this side only fills.

Runs in the reader thread; same visitor interface as @collector, so the wire dispatches to it identically.

Instance Method Summary collapse

Constructor Details

#initialize(buffer) ⇒ StreamHandler

Returns a new instance of StreamHandler.



19
20
21
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 19

def initialize(buffer)
  @buffer = buffer
end

Instance Method Details

#fail(error) ⇒ Object

Connection failure fan-out (see Connection#fan_out / Wire#fail_pending): surface the error to a cursor parked in the buffer.



47
48
49
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 47

def fail(error)
  @buffer.fail(error)
end

#on_failure(message) ⇒ Object 



35
36
37
38
39
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 35

def on_failure(message)
  # Records already delivered stay readable; the cursor hits this error
  # once it drains them (records that preceded the failure are valid).
  @buffer.fail(message.to_exception)
end

#on_ignored(_message) ⇒ Object 



41
42
43
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 41

def on_ignored(_message)
  @buffer.finish
end

#on_record(message) ⇒ Object 



23
24
25
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 23

def on_record(message)
  @buffer.push_record(message)
end

#on_success(message) ⇒ Object 



27
28
29
30
31
32
33
 
# File 'lib/neo4j/driver/bolt/stream_handler.rb', line 27

def on_success(message)
  if message.metadata[:has_more]
    @buffer.batch_complete(has_more: true)
  else
    @buffer.finish(message.metadata)
  end
end