Module: RubyReactor::Storage::RedisPubSub

Included in:
RedisAdapter
Defined in:
lib/ruby_reactor/storage/redis_pub_sub.rb

Overview

The completion-signal channel behind the notified wait. Pure latency optimisation: at-most-once, unpersisted, never load-bearing — every waiting path ends at a durable record, so a lost signal costs a fallback interval and never correctness.

Instance Method Summary collapse

Instance Method Details

#publish(channel, message) ⇒ Object 



26
27
28
 
# File 'lib/ruby_reactor/storage/redis_pub_sub.rb', line 26

def publish(channel, message)
  @redis.publish(channel, message)
end

#subscribe(channel, &block) ⇒ Object

SUBSCRIBE puts a connection into subscriber mode — every other command on it then fails — so this MUST NOT use the shared client, or one waiter would poison storage for the whole process. A dedicated connection is opened per subscription and closed on the way out.

Blocks the calling thread until the block returns truthy for a message (completion signals are one-shot) or the thread is killed.



17
18
19
20
21
22
23
24
 
# File 'lib/ruby_reactor/storage/redis_pub_sub.rb', line 17

def subscribe(channel, &block)
  connection = Redis.new(@redis_config)
  connection.subscribe(channel) do |on|
    on.message { |_channel, message| connection.unsubscribe if block.call(message) }
  end
ensure
  connection&.close
end