Class: ActsAsTbackend::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/acts_as_tbackend/connection.rb

Overview

One persistent framed connection to a TBackend daemon. Protocol parity with the canonical P16 client (token, write_fact_once, rich status mapping, Unavailable/Unknown split) — but the socket is kept open and reused across requests (reconnect only on error), which is what makes pooled throughput cheap.

NOT thread-safe: one in-flight request per connection. Concurrency is the Pool's job — check a Connection out, use it, check it back in.

Defined Under Namespace

Classes: InvalidFrame, TransportUnavailable, TransportUnknown

Constant Summary collapse

DEFAULT_HOST = 
"127.0.0.1"
DEFAULT_PORT = 
7401
MAX_FRAME_BYTES = 
64 * 1024 * 1024

Instance Method Summary collapse

Constructor Details

#initialize(host: DEFAULT_HOST, port: DEFAULT_PORT, token: nil, connect_timeout: 1.0, request_timeout: 2.0, durability_default: "accepted", strict: false) ⇒ Connection

Returns a new instance of Connection.



25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/acts_as_tbackend/connection.rb', line 25

def initialize(host: DEFAULT_HOST, port: DEFAULT_PORT, token: nil,
               connect_timeout: 1.0, request_timeout: 2.0,
               durability_default: "accepted", strict: false)
  @host = host
  @port = Integer(port)
  @token = token
  @connect_timeout = Float(connect_timeout)
  @request_timeout = Float(request_timeout)
  @durability_default = durability_default
  @strict = strict
  @socket = nil
end

Instance Method Details

#close ⇒ Object 



85
86
87
88
89
90
91
 
# File 'lib/acts_as_tbackend/connection.rb', line 85

def close
  @socket&.close
rescue StandardError
  nil
ensure
  @socket = nil
end

#facts_by_seq(store:, after_seq: 0, until_seq: nil, timeout: nil) ⇒ Object

Clock-free ordered read (the ordering authority). Prefer this over timestamp ordering for replay/audit/pull.



79
80
81
82
83
 
# File 'lib/acts_as_tbackend/connection.rb', line 79

def facts_by_seq(store:, after_seq: 0, until_seq: nil, timeout: nil)
  req = { op: "facts_by_seq", store: store.to_s, after_seq: after_seq }
  req[:until_seq] = until_seq unless until_seq.nil?
  map_generic(request(req, timeout))
end

#facts_for(store:, key: nil, since: nil, as_of: nil, timeout: nil) ⇒ Object 



69
70
71
72
73
74
75
 
# File 'lib/acts_as_tbackend/connection.rb', line 69

def facts_for(store:, key: nil, since: nil, as_of: nil, timeout: nil)
  req = { op: "facts_for", store: store.to_s }
  req[:key] = key.to_s unless key.nil?
  req[:since] = since unless since.nil?
  req[:as_of] = as_of unless as_of.nil?
  map_generic(request(req, timeout))
end

#latest_for(store:, key:, as_of: nil, timeout: nil) ⇒ Object 



63
64
65
66
67
 
# File 'lib/acts_as_tbackend/connection.rb', line 63

def latest_for(store:, key:, as_of: nil, timeout: nil)
  req = { op: "latest_for", store: store.to_s, key: key.to_s }
  req[:as_of] = as_of unless as_of.nil?
  map_generic(request(req, timeout))
end

#ping(timeout: nil) ⇒ Object 



38
39
40
 
# File 'lib/acts_as_tbackend/connection.rb', line 38

def ping(timeout: nil)
  map_generic(request({ op: "ping" }, timeout))
end

#write_fact_once(fact, durability: nil, timeout: nil) ⇒ Object

Idempotent durable write — the recommended write path. Derive fact["id"] deterministically (see Fact.derive_id) so a retry is a replay, not a duplicate.



44
45
46
47
 
# File 'lib/acts_as_tbackend/connection.rb', line 44

def write_fact_once(fact, durability: nil, timeout: nil)
  req = { op: "write_fact_once", fact: fact, durability: durability || @durability_default }
  map_write_once(request(req, timeout))
end

#write_fact_once_safe(fact, durability: nil, timeout: nil, attempts: 2, backoff: 0.05) ⇒ Object

Bounded retry of write_fact_once for the retry-safe transient states (rejected_before_commit / timeout_unknown). Never loops unbounded.



51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/acts_as_tbackend/connection.rb', line 51

def write_fact_once_safe(fact, durability: nil, timeout: nil, attempts: 2, backoff: 0.05)
  max = [Integer(attempts), 1].max
  seen = []
  max.times do |i|
    result = write_fact_once(fact, durability: durability, timeout: timeout)
    seen << summary(result)
    return result.merge(attempt_count: seen.length, attempts: seen) unless retry_safe?(result[:status]) && i + 1 < max

    sleep(backoff.to_f) if backoff.to_f.positive?
  end
end