Module: PatientHttp::SolidQueue

Defined in:
lib/patient_http/solid_queue.rb,
lib/patient_http/solid_queue/engine.rb,
lib/patient_http/solid_queue/record.rb,
lib/patient_http/solid_queue/context.rb,
lib/patient_http/solid_queue/gc_lock.rb,
lib/patient_http/solid_queue/request_job.rb,
lib/patient_http/solid_queue/callback_job.rb,
lib/patient_http/solid_queue/task_handler.rb,
lib/patient_http/solid_queue/task_monitor.rb,
lib/patient_http/solid_queue/configuration.rb,
lib/patient_http/solid_queue/lifecycle_hooks.rb,
lib/patient_http/solid_queue/inflight_request.rb,
lib/patient_http/solid_queue/request_executor.rb,
lib/patient_http/solid_queue/processor_observer.rb,
lib/patient_http/solid_queue/task_monitor_thread.rb,
lib/patient_http/solid_queue/process_registration.rb

Defined Under Namespace

Classes: CallbackJob, Configuration, Context, Engine, GcLock, InflightRequest, LifecycleHooks, ProcessRegistration, ProcessorObserver, Record, RegistrationError, RequestExecutor, RequestJob, TaskHandler, TaskMonitor, TaskMonitorThread

Constant Summary collapse

VERSION = 
File.read(File.join(__dir__, "../../VERSION")).strip

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.configuration ⇒ Configuration

Return the current configuration, initializing with defaults if necessary.

Returns:



90
91
92
 
# File 'lib/patient_http/solid_queue.rb', line 90

def configuration
  @configuration ||= Configuration.new
end

Class Method Details

.after_completion {|response| ... } ⇒ Object

Add a callback to be executed after a successful request completion.

Yields:

  • (response) —

    block to execute after an HTTP request completes

Yield Parameters:

  • response (PatientHttp::Response) —

    the HTTP response



107
108
109
 
# File 'lib/patient_http/solid_queue.rb', line 107

def after_completion(&block)
  @after_completion_callbacks << block
end

.after_error {|error| ... } ⇒ Object

Add a callback to be executed after a request error.

Yields:

  • (error) —

    block to execute after an HTTP request errors

Yield Parameters:

  • error (PatientHttp::Error) —

    information about the error



115
116
117
 
# File 'lib/patient_http/solid_queue.rb', line 115

def after_error(&block)
  @after_error_callbacks << block
end

.configure {|Configuration| ... } ⇒ Configuration

Configure the gem with a block. The built configuration is also set as the PatientHttp.default_configuration so that secrets registered at the module level with PatientHttp.register_secret are applied to the configuration the processor runs with, regardless of boot order.

Yields:

Returns:



77
78
79
80
81
82
83
84
85
 
# File 'lib/patient_http/solid_queue.rb', line 77

def configure
  configuration = Configuration.new
  yield(configuration) if block_given?
  @configuration = configuration
  @external_storage = nil
  register_handler
  PatientHttp.default_configuration = configuration
  configuration
end

.decrypt(value) ⇒ Object

Decrypt a value using the configured encryptor.

Parameters:

  • value (String) —

    the encrypted value to decrypt

Returns:

  • (Object) —

    the decrypted value



332
333
334
 
# File 'lib/patient_http/solid_queue.rb', line 332

def decrypt(value)
  configuration.encryptor.decrypt(value)
end

.draining? ⇒ Boolean

Check if any processor is draining (not accepting new requests).

Returns:

  • (Boolean) 


129
130
131
 
# File 'lib/patient_http/solid_queue.rb', line 129

def draining?
  @processors.values.any?(&:draining?)
end

.encrypt(value) ⇒ String

Encrypt a value using the configured encryptor.

Parameters:

  • value (Object) —

    the value to encrypt

Returns:

  • (String) —

    the encrypted value



324
325
326
 
# File 'lib/patient_http/solid_queue.rb', line 324

def encrypt(value)
  configuration.encryptor.encrypt(value)
end

.execute(request, callback:, callback_args: nil, raise_error_responses: false, processor: nil) ⇒ String

Execute an async HTTP request.

Parameters:

  • request (PatientHttp::Request) —

    the HTTP request to execute

  • callback (Class, String) —

    Callback service class with on_complete and on_error instance methods, or its fully qualified class name.

  • callback_args (#to_h, nil) (defaults to: nil) —

    Arguments to pass to callback

  • raise_error_responses (Boolean) (defaults to: false) —

    If true, treats non-2xx responses as errors

  • processor (Symbol, String, nil) (defaults to: nil) —

    Name of the processor profile that should execute the request. Defaults to the request's own processor name or :default.

Returns:

  • (String) —

    the request ID

Raises:

  • (PatientHttp::UnknownProcessorError) —

    if the processor profile is not configured



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/patient_http/solid_queue.rb', line 166

def execute(request, callback:, callback_args: nil, raise_error_responses: false, processor: nil)
  PatientHttp::CallbackValidator.validate!(callback)
  callback_name = callback.is_a?(Class) ? callback.name : callback.to_s
  callback_args = PatientHttp::CallbackValidator.validate_callback_args(callback_args)
  request_id = SecureRandom.uuid
  processor_name = (processor || request.processor || :default).to_s

  # Catch a misspelled profile name at the call site. A job that names an
  # unconfigured profile is retried instead, which covers rolling deploys
  # where the executing process is older than the enqueueing one.
  unless configuration.processor_profiles.key?(processor_name.to_sym)
    raise PatientHttp::UnknownProcessorError.new("No processor profile configured for #{processor_name.inspect}")
  end

  encrypted = encrypt(request.as_json)

  data = if external_storage.enabled?
    external_storage.store(encrypted, max_size: configuration.payload_store_threshold)
  else
    encrypted
  end

  RequestJob.perform_later(data, callback_name, raise_error_responses, callback_args, request_id, processor_name)

  request_id
end

.external_storage ⇒ PatientHttp::ExternalStorage

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get an ExternalStorage instance for storing and fetching payloads.

Returns:

  • (PatientHttp::ExternalStorage) 


151
152
153
 
# File 'lib/patient_http/solid_queue.rb', line 151

def external_storage
  @external_storage ||= PatientHttp::ExternalStorage.new(configuration)
end

.invoke_completion_callbacks(response) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Invoke the registered completion callbacks.

Parameters:

  • response (PatientHttp::Response) —

    the HTTP response



299
300
301
302
303
304
305
 
# File 'lib/patient_http/solid_queue.rb', line 299

def invoke_completion_callbacks(response)
  @after_completion_callbacks.each do |callback|
    callback.call(response)
  rescue => e
    configuration.logger&.error("[PatientHttp::SolidQueue] after_completion callback error: #{e.class} - #{e.message}")
  end
end

.invoke_error_callbacks(error) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Invoke the registered error callbacks.

Parameters:

  • error (PatientHttp::Error) —

    information about the error



312
313
314
315
316
317
318
 
# File 'lib/patient_http/solid_queue.rb', line 312

def invoke_error_callbacks(error)
  @after_error_callbacks.each do |callback|
    callback.call(error)
  rescue => e
    configuration.logger&.error("[PatientHttp::SolidQueue] after_error callback error: #{e.class} - #{e.message}")
  end
end

.processor(name = :default) ⇒ PatientHttp::Processor?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a processor instance by name (internal accessor).

Parameters:

  • name (Symbol, String) (defaults to: :default) —

    the processor name

Returns:

  • (PatientHttp::Processor, nil) 


341
342
343
 
# File 'lib/patient_http/solid_queue.rb', line 341

def processor(name = :default)
  @processors[name.to_sym]
end

.processor=(value) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Set the default processor (internal, for testing).

Parameters:

  • value (PatientHttp::Processor, nil) 


349
350
351
352
353
354
355
 
# File 'lib/patient_http/solid_queue.rb', line 349

def processor=(value)
  if value.nil?
    @processors.delete(:default)
  else
    @processors[:default] = value
  end
end

.quiet ⇒ void

This method returns an undefined value.

Signal all processors to drain (stop accepting new requests).



232
233
234
235
236
237
238
 
# File 'lib/patient_http/solid_queue.rb', line 232

def quiet
  @lifecycle_mutex.synchronize do
    return unless running?

    @processors.each_value(&:drain)
  end
end

.register_handler ⇒ void

This method returns an undefined value.

Register SolidQueue as the request handler for processing HTTP requests. This is called automatically when the processor starts or you call PatientHttp::SolidQueue.configure.



281
282
283
284
285
286
287
288
289
290
291
292
 
# File 'lib/patient_http/solid_queue.rb', line 281

def register_handler
  @request_handler ||= lambda do |request:, callback:, raise_error_responses:, callback_args:|
    execute(
      request,
      callback: callback,
      raise_error_responses: raise_error_responses,
      callback_args: callback_args
    )
  end

  PatientHttp.register_handler(@request_handler)
end

.reset! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Reset all state (useful for testing).



262
263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/patient_http/solid_queue.rb', line 262

def reset!
  if @request_handler
    PatientHttp.unregister_handler(@request_handler)
  end
  @lifecycle_mutex.synchronize do
    stop_processors(timeout: 0)
    @processors = {}
    shutdown_shared_services
  end
  @configuration = nil
  @external_storage = nil
  @after_completion_callbacks = []
  @after_error_callbacks = []
end

.reset_configuration! ⇒ Configuration

Reset configuration to defaults (useful for testing).

Returns:



97
98
99
100
101
 
# File 'lib/patient_http/solid_queue.rb', line 97

def reset_configuration!
  @configuration = nil
  @external_storage = nil
  configuration
end

.running? ⇒ Boolean

Check if any processor is running.

Returns:

  • (Boolean) 


122
123
124
 
# File 'lib/patient_http/solid_queue.rb', line 122

def running?
  @processors.values.any?(&:running?)
end

.start ⇒ void

This method returns an undefined value.

Start a processor for each configured processor profile, along with the shared crash-recovery monitor.



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/patient_http/solid_queue.rb', line 197

def start
  @lifecycle_mutex.synchronize do
    return if @processors.any? && !@processors.values.all?(&:stopped?)

    @task_monitor ||= TaskMonitor.new(
      configuration,
      max_connections: -> { @processors.values.sum { |p| p.config.max_connections } }
    )

    @processors = {}
    configuration.processor_profiles.each_key do |name|
      processor = PatientHttp::Processor.new(configuration.processor_config(name), name: name)
      processor.observe(ProcessorObserver.new(processor, task_monitor: @task_monitor))
      @processors[name] = processor
    end
    @processors.each_value(&:start)

    # A previous run can leave a monitor thread behind if the processors
    # stopped without going through #stop.
    @monitor_thread&.stop

    @monitor_thread = TaskMonitorThread.new(
      configuration,
      @task_monitor,
      -> { @processors.values.flat_map(&:tracked_request_ids) }
    )
    @monitor_thread.start
  end

  register_handler
end

.stop(timeout: nil) ⇒ void

This method returns an undefined value.

Stop all processors gracefully.

Parameters:

  • timeout (Float, nil) (defaults to: nil) —

    maximum time to wait for in-flight requests to complete



244
245
246
247
248
249
250
251
252
253
254
255
256
 
# File 'lib/patient_http/solid_queue.rb', line 244

def stop(timeout: nil)
  if @request_handler
    PatientHttp.unregister_handler(@request_handler)
  end

  @lifecycle_mutex.synchronize do
    return if @processors.empty?

    stop_processors(timeout: timeout)
    @processors = {}
    shutdown_shared_services
  end
end

.stopped? ⇒ Boolean

Check if all processors are stopped or none have been started.

Returns:

  • (Boolean) 


143
144
145
 
# File 'lib/patient_http/solid_queue.rb', line 143

def stopped?
  @processors.values.all?(&:stopped?)
end

.stopping? ⇒ Boolean

Check if any processor is stopping.

Returns:

  • (Boolean) 


136
137
138
 
# File 'lib/patient_http/solid_queue.rb', line 136

def stopping?
  @processors.values.any?(&:stopping?)
end