Class: Pgbus::Web::Streamer::MasterHubBoot

Inherits:
Object
  • Object
show all
Defined in:
lib/pgbus/web/streamer/master_hub_boot.rb

Overview

Deferred MasterHub startup for the Puma master (issue #382). The pgbus_streams plugin's start runs BEFORE preload_app! loads the Rails app (and with it the pgbus initializer), so the hub cannot be built eagerly. This class splits the two halves:

1. The socket path is exported to ENV IMMEDIATELY — workers inherit
 it across fork and connect lazily on first SSE use.
2. A poller thread waits for Pgbus.configuration to become ready
 (the initializer has run — with preload_app!, before the first
 fork), then builds and starts the MasterHub. Workers that race a
 still-booting hub simply fail to connect and fall back to their
 own listener until they recycle — degraded footprint, never
 degraded semantics.

Without preload_app! the master never loads the app, the deadline expires quietly, no socket is ever bound, and every worker keeps today's per-worker listener — :master scope effectively requires preload_app!, documented on the docs site.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket_path: self.class.default_socket_path, hub_factory: nil, poll_interval: 1.0, deadline: 120, logger: nil) ⇒ MasterHubBoot

Returns a new instance of MasterHubBoot.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/pgbus/web/streamer/master_hub_boot.rb', line 31

def initialize(socket_path: self.class.default_socket_path, hub_factory: nil,
               poll_interval: 1.0, deadline: 120, logger: nil)
  @socket_path = socket_path
  @hub_factory = hub_factory || lambda do |socket_path:|
    MasterHub.new(config: Pgbus.configuration, socket_path: socket_path)
  end
  @poll_interval = poll_interval
  @deadline = deadline
  @logger = logger
  # Guards @hub and @running: written by the caller thread
  # (start/stop) and the background poller. A hub whose start
  # outlives stop's join budget is stopped by whichever side sees
  # the flag last, so teardown can never leave a live hub behind.
  @state_mutex = Mutex.new
  @hub = nil
  @running = false
  @thread = nil
end

Class Method Details

.default_socket_path ⇒ Object 



27
28
29
 
# File 'lib/pgbus/web/streamer/master_hub_boot.rb', line 27

def self.default_socket_path
  File.join(Dir.tmpdir, "pgbus-streams-hub-#{::Process.pid}.sock")
end

Instance Method Details

#start ⇒ Object 



50
51
52
53
54
55
 
# File 'lib/pgbus/web/streamer/master_hub_boot.rb', line 50

def start
  ENV["PGBUS_STREAMS_HUB_SOCKET"] = @socket_path
  @state_mutex.synchronize { @running = true }
  @thread = Thread.new { wait_and_start }
  self
end

#stop ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/pgbus/web/streamer/master_hub_boot.rb', line 57

def stop
  to_stop = @state_mutex.synchronize do
    @running = false
    hub = @hub
    @hub = nil
    hub
  end
  @thread&.join(2)
  @thread = nil
  to_stop&.stop
  self
end