Class: Neo4j::Driver::Bolt::Wire

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

Overview

Sans-I/O Bolt protocol core: a pure state machine with no socket.

* `enqueue(message)` packs + chunk-frames a request into an outbound
byte buffer; `take_outbound` hands those bytes to whoever owns the
socket.
* `receive(bytes)` feeds inbound bytes and returns the complete
messages now decodable (Success / Failure / Record / Ignored), in
order — reassembling chunks across calls, skipping NOOP keepalives,
and retaining any trailing partial bytes for the next call.

Because it never touches a socket, the whole Bolt framing + hydration path is unit-testable without a network, and stays identical regardless of how I/O is driven (blocking thread, fiber pump, on-demand). See docs/sans-io-pump.md.

Created after the handshake — it needs the negotiated protocol to configure the packer (UTC datetime flag) and to customise hydration for version-specific message shapes.

Constant Summary collapse

END_MARKER =

Bolt message chunking: a message is a sequence of [u16 size][size bytes] chunks terminated by a zero-size chunk (the 0x00 0x00 end marker). A bare end marker with no preceding chunks is a NOOP — an inline keepalive the server sends to keep a slow response under the recv-timeout; it carries no message.

"\x00\x00".b
MAX_CHUNK = 
65_535

Instance Method Summary collapse

Constructor Details

#initialize(protocol) ⇒ Wire

Returns a new instance of Wire.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 33

def initialize(protocol)
  @protocol = protocol
  @packer = PackStream::Packer.new(@protocol)
  @protocol.configure_packer(@packer)
  @outbound = binary_string # consumer-only (enqueue/take_outbound)
  @inbound = binary_string # reader-only: received bytes not yet parsed
  @message = binary_string # reader-only: chunks of the in-progress message
  # Response-ordering FIFO. Bolt replies in request order, so each sent
  # request pushes the handler that will receive its response(s); the
  # front handler always owns the next reply. A handler is any response
  # visitor (responds to on_record/on_success/on_failure/on_ignored —
  # the same interface Message#accept dispatches to).
  #
  # This FIFO is the one piece of wire state touched by two threads: the
  # consumer pushes (enqueue) while the dedicated reader shifts/dispatches
  # (receive) and may clear it on failure (fail_pending). @mutex serialises
  # those — without it JRuby (no GVL) races the Array and loses handlers,
  # which strands a consumer parked on a buffer that never fills. Handler
  # delivery is non-blocking (records/replies land on unbounded queues —
  # flow control is the cursor's watermark, not a bounded buffer), so
  # dispatching under the lock can't stall the reader.
  @handlers = []
  @mutex = Mutex.new
end

Instance Method Details

#enable_utc_datetime ⇒ Object

Switch datetime packing to UTC-seconds (0x49/0x69). Connection calls this when the Bolt 4.3/4.4 HELLO confirms patch_bolt: ["utc"]; on 5.0+ configure_packer already set it, so this only fires on 4.x.



61
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 61

def enable_utc_datetime = @packer.use_utc_datetime = true

#enqueue(message, handler) ⇒ Object

Pack + chunk-frame message into the outbound buffer and register the handler for its reply. Several enqueues before a take_outbound are exactly the pipelining Bolt expects (HELLO+LOGON, RUN+PULL); the FIFO keeps each request matched to its response.



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 67

def enqueue(message, handler)
  @packer.reset
  @packer.pack_message(message)
  data = @packer.bytes

  offset = 0
  while offset < data.bytesize
    size = [data.bytesize - offset, MAX_CHUNK].min
    @outbound << [size].pack('S>') << data.byteslice(offset, size)
    offset += size
  end
  @outbound << END_MARKER
  @mutex.synchronize { @handlers.push(handler) }
  self
end

#fail_pending(error) ⇒ Object

Connection failure fan-out: tell every outstanding request's handler the connection died, so a consumer parked on that handler's buffer (or completion) wakes and re-raises rather than hanging. Drops them from the FIFO — the wire is done.



121
122
123
124
125
126
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 121

def fail_pending(error)
  @mutex.synchronize do
    @handlers.each { |handler| handler.fail(error) }
    @handlers.clear
  end
end

#in_flight ⇒ Object

Requests still awaiting their terminal reply.



86
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 86

def in_flight = @mutex.synchronize { @handlers.size }

#pending_outbound? ⇒ Boolean

Returns:

  • (Boolean) 


83
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 83

def pending_outbound? = !@outbound.empty?

#receive(bytes) ⇒ Object

Feed inbound bytes and route each fully-decoded message to the front handler (via its own #accept visitor). A RECORD keeps the handler at the front (one request streams many); a terminal (SUCCESS/FAILURE/ IGNORED) completes the request and pops it. NOOP keepalives carry no message and are skipped. Partial bytes are retained for the next call.



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 100

def receive(bytes)
  @inbound << bytes
  # @inbound/@message/next_message are reader-only; only the @handlers
  # touch needs the lock. Dispatch (accept) happens under it too — that's
  # safe because delivery is non-blocking — and shifting after accept keeps
  # in_flight from hitting zero until the terminal has actually been
  # delivered (the quiescence contract reset!/drain_quiesced rely on).
  @mutex.synchronize do
    while (message = next_message)
      next if message == :noop || @handlers.empty?

      message.accept(@handlers.first)
      @handlers.shift if message.terminal?
    end
  end
end

#take_outbound ⇒ Object

Hand over (and clear) the framed bytes to write to the socket.



89
90
91
92
93
 
# File 'lib/neo4j/driver/bolt/wire.rb', line 89

def take_outbound
  out = @outbound
  @outbound = binary_string
  out
end