Class: Hop::Endpoint

Inherits:
Object
  • Object
show all
Defined in:
lib/hop/endpoint.rb

Overview

Receive Hop messages in Ruby with a Sinatra/Rails-shaped surface, over the libhop C ABI.

hop = Hop::Endpoint.new
hop.on("acme/orders") { |req, reply| reply.call(201, "ok") }  # req.from is VERIFIED

Semantics: inbound is a durable store-and-forward consume; a reply is a new addressed message that may arrive later. The DX is HTTP-shaped; delivery is delay-tolerant. core is poll-model, so the endpoint runs a background pump thread (the node is thread-safe).

Constant Summary collapse

CLOSED =

Sentinel pushed to a pending request queue by #close, so a blocked caller fails fast instead of waiting out its full timeout. A unique object, never equal to a real [status, body] response.

Object.new

Instance Method Summary collapse

Constructor Details

#initialize(key: nil, tick_ms: 50, cluster: nil, quorum: nil) ⇒ Endpoint

Returns a new instance of Endpoint.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/hop/endpoint.rb', line 26

def initialize(key: nil, tick_ms: 50, cluster: nil, quorum: nil)
  Hop::FFI.assert_abi!
  @node = key ? Hop::FFI.node_with_secret(key) : Hop::FFI.node_new
  Hop::FFI.tick(@node, now_ms)
  Hop::FFI.publish_prekey(@node)
  @handlers = {}
  @links = {}
  @pending = {}
  @closers = []
  @mutex = Mutex.new         # guards @pending
  @node_lock = Monitor.new   # serializes every libhop call on @node vs. #close; reentrant so a
  @closed = false            # reply issued from inside #pump re-enters without deadlocking
  cluster(cluster) if cluster # dedup across sibling replicas (same identity, no shared store)
  cluster_quorum(quorum) if quorum # TTL-based visibility threshold; not consensus
  @thread = Thread.new { pump_loop(tick_ms / 1000.0) }
end

Instance Method Details

#address ⇒ Object 



43
 
# File 'lib/hop/endpoint.rb', line 43

def address = Hop::FFI.to_b58(address_bytes)

#address_bytes ⇒ Object 



44
 
# File 'lib/hop/endpoint.rb', line 44

def address_bytes = with_node { |n| Hop::FFI.address(n) }

#attach(port, ssl_context, public_url, ttl_secs: 3600) ⇒ Object

Start an HTTPS server (WSS bearer at /_hop + /.well-known/hop) IN ONE CALL. public_url is where senders reach it, e.g. "wss://myaddress.com/_hop". Returns the server (call #shutdown to stop).



107
108
109
110
 
# File 'lib/hop/endpoint.rb', line 107

def attach(port, ssl_context, public_url, ttl_secs: 3600)
  require "hop/wss_bearer"
  Hop::WssBearer.serve(self, port, ssl_context, public_url, ttl_secs)
end

#close ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/hop/endpoint.rb', line 146

def close
  @node_lock.synchronize do
    return if @closed

    @closed = true
  end
  @closers.each { |c| c.call rescue nil } # unblock bearer accept/read threads so they exit
  # Wake any in-flight request waiters so they fail fast instead of blocking their full timeout.
  @mutex.synchronize do
    @pending.each_value { |q| q.push(CLOSED) }
    @pending.clear
  end
  @thread.join(1) unless Thread.current == @thread
  # Free only after @closed is set and the pump has stopped: a late bearer-thread call (a WSS
  # run_link firing #link_down as its socket EOFs) now short-circuits in #with_node instead of
  # dereferencing a freed node.
  @node_lock.synchronize do
    next unless @node

    Hop::FFI.node_free(@node)
    @node = nil
  end
end

#cluster(secret_or_passphrase) ⇒ Object

Join the endpoint cluster so sibling replicas (same identity, no shared datastore) each handle a given request once. Pass a String passphrase (interops with the service's HOP_CLUSTER_SECRET) or a 32-byte binary String secret. Dedup then applies transparently. Returns self.



49
50
51
52
53
54
55
56
57
58
 
# File 'lib/hop/endpoint.rb', line 49

def cluster(secret_or_passphrase)
  if secret_or_passphrase.is_a?(String) && secret_or_passphrase.encoding != Encoding::BINARY
    with_node { |n| Hop::FFI.cluster_join_passphrase(n, secret_or_passphrase) }
  else
    b = secret_or_passphrase.to_str
    raise ArgumentError, "cluster secret must be 32 bytes or a passphrase string" unless b.bytesize == 32
    with_node { |n| Hop::FFI.cluster_join(n, b) }
  end
  self
end

#cluster_members ⇒ Object

Live replica count (self + peers within the membership TTL); 1 if not clustered.



61
 
# File 'lib/hop/endpoint.rb', line 61

def cluster_members = with_node { |n| Hop::FFI.cluster_members(n) }

#cluster_quorum(min) ⇒ Object

Require at least min live cluster members visible before this replica will process a request using a TTL-based visibility threshold. This is a conservative failover heuristic, not consensus or an at-most-once guarantee. 0 or 1 disables it. Returns self.



66
67
68
69
 
# File 'lib/hop/endpoint.rb', line 66

def cluster_quorum(min)
  with_node { |n| Hop::FFI.cluster_set_quorum(n, min) }
  self
end

#deliver(link, data) ⇒ Object 



137
 
# File 'lib/hop/endpoint.rb', line 137

def deliver(link, data) = with_node { |n| Hop::FFI.received(n, link, data) }

#dial_by_name(base_url, insecure_tls: false) ⇒ Object

Resolve a base HTTPS URL to a verified endpoint, dial its WSS, and return the reachable address (then use #request). Set insecure_tls: true only for a dev/self-signed cert.



114
115
116
117
118
119
120
 
# File 'lib/hop/endpoint.rb', line 114

def dial_by_name(base_url, insecure_tls: false)
  require "hop/discovery"
  require "hop/wss_bearer"
  info = Hop::Discovery.resolve(base_url, insecure_tls: insecure_tls)
  Hop::WssBearer.dial(self, info[:wss_url], insecure_tls: insecure_tls)
  info[:address]
end

#link_down(link) ⇒ Object 



139
140
141
142
143
144
 
# File 'lib/hop/endpoint.rb', line 139

def link_down(link)
  with_node do |n|
    @links.delete(link)
    Hop::FFI.disconnected(n, link)
  end
end

#on(service, &block) ⇒ Object

Register a receiver for a hops:// service. The block gets (req, reply); reply is a callable reply.call(status, body).



73
74
75
76
77
 
# File 'lib/hop/endpoint.rb', line 73

def on(service, &block)
  with_node { |n| Hop::FFI.subscribe(n, service) }
  @handlers[service] = block
  self
end

#register_closer(&block) ⇒ Object

Register a teardown hook (e.g. a bearer's listening socket). #close runs these before freeing the node so bearer threads unblock and exit. If already closed, the hook fires immediately.



124
125
126
127
 
# File 'lib/hop/endpoint.rb', line 124

def register_closer(&block)
  run_now = @node_lock.synchronize { @closed ? true : (@closers << block; false) }
  block.call if run_now
end

#register_link(link, role, send_fn) ⇒ Object

---- bearer seam (called from bearer threads) ----



130
131
132
133
134
135
 
# File 'lib/hop/endpoint.rb', line 130

def register_link(link, role, send_fn)
  with_node do |n|
    @links[link] = send_fn
    Hop::FFI.connected(n, link, role == :dialer)
  end
end

#request(dst, service, method, args = "", timeout: 15.0) ⇒ Object

Call a service on a remote endpoint. Blocks until the response returns (delay-tolerant).



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/hop/endpoint.rb', line 80

def request(dst, service, method, args = "", timeout: 15.0)
  dst_bytes = dst.is_a?(String) && dst.bytesize == 32 ? dst : Hop::FFI.from_b58(dst)
  q = Queue.new
  # Send and register the waiter atomically under @node_lock so #pump (which also holds it) cannot
  # deliver the response before @pending knows to route it.
  req_id = with_node do |n|
    id = Hop::FFI.send_service_request(n, dst_bytes, service, method, to_bytes(args))
    @mutex.synchronize { @pending[id] = q }
    id
  end
  raise "endpoint is closed" unless req_id

  begin
    res = Timeout.timeout(timeout) { q.pop } # [status, body], or CLOSED if #close woke us
    raise "endpoint is closed" if res == CLOSED
    res
  rescue Timeout::Error
    @mutex.synchronize { @pending.delete(req_id) }
    raise "hops://#{service}/#{method} timed out after #{timeout}s"
  end
end

#sign_reach(endpoint, ttl_secs = 3600) ⇒ Object

Sign a self-certifying reachability record for this endpoint's address bound to endpoint.



103
 
# File 'lib/hop/endpoint.rb', line 103

def sign_reach(endpoint, ttl_secs = 3600) = with_node { |n| Hop::FFI.sign_reach(n, endpoint, ttl_secs) }