Class: SolidObjects::WakeUpAdapters::Redis

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_objects/wake_up_adapters/redis.rb,
sig/generated/lib/solid_objects/wake_up_adapters/redis.rbs

Overview

Wakes runtime roles across processes using Redis publish/subscribe.

MySQL has no notification primitive, so this is the cross-process option for applications that cannot use PostgreSQL notifications. It is optional in every sense: the redis gem is not a dependency of this gem, and the polling interval remains the upper bound, so a missed or failed notification costs latency rather than correctness.

Constant Summary collapse

CHANNEL =

Returns:

  • (::String) 
"solid_objects_wake_up"
FAILED_WAIT_INTERVAL =

Returns:

  • (::Float) 
0.05
SUBSCRIBE_TIMEOUT =

Returns:

  • (::Float) 
5.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(channel: CHANNEL, url: nil, client: nil) ⇒ Redis

Returns a new instance of Redis.

RBS:

  • (?channel: String, ?url: String?, ?client: untyped) -> void

Parameters:

  • channel: (String) (defaults to: CHANNEL)
  • url: (String, nil) (defaults to: nil)
  • client: (Object) (defaults to: nil) 


31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 31

def initialize(channel: CHANNEL, url: nil, client: nil)
  @channel = channel
  @url = url
  @client = client
  @mutex = Thread::Mutex.new
  @condition = Thread::ConditionVariable.new
  @subscriber = nil
  @subscription = nil
  @signalled = 0
  validate_client!
end

Instance Attribute Details

#channel ⇒ Object (readonly)

RBS:

  • @channel: String

  • @url: String?

  • @client: untyped

  • @mutex: Thread::Mutex

  • @condition: Thread::ConditionVariable

  • @subscriber: Thread?

  • @subscription: untyped

  • @signalled: Integer

Returns:

  • (Object) 


28
29
30
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 28

def channel
  @channel
end

#condition ⇒ Object (readonly)

Returns the value of attribute condition.

Returns:

  • (Object) 


104
105
106
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 104

def condition
  @condition
end

#mutex ⇒ Object (readonly)

Returns the value of attribute mutex.

Returns:

  • (Object) 


104
105
106
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 104

def mutex
  @mutex
end

#url ⇒ Object (readonly)

Returns the value of attribute url.

Returns:

  • (Object) 


104
105
106
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 104

def url
  @url
end

Instance Method Details

#broadcast ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



120
121
122
123
124
125
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 120

def broadcast
  mutex.synchronize do
    @signalled += 1
    condition.broadcast
  end
end

#build_client ⇒ Object

RBS:

  • () -> untyped

Returns:

  • (Object) 


139
140
141
142
143
144
145
146
147
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 139

def build_client
  return @client.call if @client.respond_to?(:call)

  require "redis"
  url ? ::Redis.new(url:) : ::Redis.new
rescue LoadError
  raise ArgumentError,
    "the redis gem is required for SolidObjects::WakeUpAdapters::Redis"
end

#disconnect(connection) ⇒ void

This method returns an undefined value.

RBS:

  • (untyped) -> void

Parameters:

  • (Object) 


157
158
159
160
161
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 157

def disconnect(connection)
  connection&.close
rescue
  nil
end

#instrument_failure(operation, error) ⇒ void

This method returns an undefined value.

RBS:

  • (Symbol, Exception) -> void

Parameters:

  • (Symbol)
  • (Exception) 


172
173
174
175
176
177
178
179
180
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 172

def instrument_failure(operation, error)
  SolidObjects.instrument(
    :"wake_up.failed",
    adapter: "redis",
    operation: operation.to_s,
    error_class: error.class.name,
    error_message: error.message
  )
end

#listen ⇒ Boolean

Redis delivers to a subscribed connection only, and a subscribed connection cannot serve other callers, so one background subscription per process fans out to every waiting role in memory. Subscribing eagerly also closes the window where a signal sent during startup would be missed.

RBS:

  • () -> bool

Returns:

  • (Boolean) 


74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 74

def listen
  mutex.synchronize do
    return true if @subscriber&.alive?

    ready = Queue.new
    @subscriber = Thread.new { subscribe_loop(ready) }
    Timeout.timeout(SUBSCRIBE_TIMEOUT) { ready.pop } == :subscribed
  end
rescue => error
  instrument_failure(:listen, error)
  false
end

#pace_after_failure(timeout) ⇒ void

This method returns an undefined value.

RBS:

  • (Numeric) -> void

Parameters:

  • (Numeric) 


164
165
166
167
168
169
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 164

def pace_after_failure(timeout)
  interval = [ timeout.to_f, FAILED_WAIT_INTERVAL ].min
  return unless interval.positive?

  sleep interval
end

#paced_failure(timeout) ⇒ Boolean

RBS:

  • (Numeric) -> bool

Parameters:

  • (Numeric)

Returns:

  • (Boolean) 


128
129
130
131
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 128

def paced_failure(timeout)
  pace_after_failure(timeout)
  false
end

#publisher ⇒ Object

RBS:

  • () -> untyped

Returns:

  • (Object) 


134
135
136
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 134

def publisher
  @publisher ||= build_client
end

#signal ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


44
45
46
47
48
49
50
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 44

def signal
  publisher.publish(channel, "1")
  true
rescue => error
  instrument_failure(:signal, error)
  false
end

#stop ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 88

def stop
  subscriber = mutex.synchronize do
    thread = @subscriber
    @subscriber = nil
    thread
  end
  return false unless subscriber

  disconnect(@subscription)
  subscriber.join(SUBSCRIBE_TIMEOUT)
  subscriber.kill if subscriber.alive?
  true
end

#subscribe_loop(ready) ⇒ void

This method returns an undefined value.

RBS:

  • (Queue) -> void

Parameters:

  • (Queue) 


107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 107

def subscribe_loop(ready)
  connection = build_client
  @subscription = connection
  connection.subscribe(channel) do |on|
    on.subscribe { ready << :subscribed }
    on.message { broadcast }
  end
rescue => error
  instrument_failure(:subscribe, error)
  ready << :failed
end

#validate_client! ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



150
151
152
153
154
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 150

def validate_client!
  return if @client.nil? || @client.respond_to?(:call)

  raise ArgumentError, "client must respond to call and return a Redis client"
end

#wait(timeout:) ⇒ Boolean

The counter is snapshotted before subscribing, and re-checked before blocking, so a signal delivered while this caller was still getting ready is observed rather than absorbed into the new baseline.

RBS:

  • (timeout: Numeric) -> bool

Parameters:

  • timeout: (Numeric)

Returns:

  • (Boolean) 


56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/solid_objects/wake_up_adapters/redis.rb', line 56

def wait(timeout:)
  signalled = mutex.synchronize { @signalled }
  return paced_failure(timeout) unless listen

  mutex.synchronize do
    return true unless @signalled == signalled

    condition.wait(mutex, timeout.to_f)
    @signalled != signalled
  end
end