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, 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:



83
84
85
 
# File 'lib/patient_http/solid_queue.rb', line 83

def configuration
  @configuration ||= Configuration.new
end

.processor ⇒ 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 the processor instance.

Returns:

  • (PatientHttp::Processor, nil) 


299
300
301
 
# File 'lib/patient_http/solid_queue.rb', line 299

def processor
  @processor
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



100
101
102
 
# File 'lib/patient_http/solid_queue.rb', line 100

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



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

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:



70
71
72
73
74
75
76
77
78
 
# File 'lib/patient_http/solid_queue.rb', line 70

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



291
292
293
 
# File 'lib/patient_http/solid_queue.rb', line 291

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

.draining? ⇒ Boolean

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

Returns:

  • (Boolean) 


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

def draining?
  !!@processor&.draining?
end

.encrypt(value) ⇒ String

Encrypt a value using the configured encryptor.

Parameters:

  • value (Object) —

    the value to encrypt

Returns:

  • (String) —

    the encrypted value



283
284
285
 
# File 'lib/patient_http/solid_queue.rb', line 283

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

.execute(request, callback:, callback_args: nil, raise_error_responses: false) ⇒ 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

Returns:

  • (String) —

    the request ID



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/patient_http/solid_queue.rb', line 156

def execute(request, callback:, callback_args: nil, raise_error_responses: false)
  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

  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)

  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) 


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

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



258
259
260
261
262
263
264
 
# File 'lib/patient_http/solid_queue.rb', line 258

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



271
272
273
274
275
276
277
 
# File 'lib/patient_http/solid_queue.rb', line 271

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

.quiet ⇒ void

This method returns an undefined value.

Signal the processor to drain (stop accepting new requests).



193
194
195
196
197
198
199
 
# File 'lib/patient_http/solid_queue.rb', line 193

def quiet
  @lifecycle_mutex.synchronize do
    return unless running?

    @processor.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.



240
241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/patient_http/solid_queue.rb', line 240

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).



222
223
224
225
226
227
228
229
230
231
232
233
234
 
# File 'lib/patient_http/solid_queue.rb', line 222

def reset!
  if @request_handler
    PatientHttp.unregister_handler(@request_handler)
  end
  @lifecycle_mutex.synchronize do
    @processor&.stop(timeout: 0)
    @processor = nil
  end
  @configuration = nil
  @external_storage = nil
  @after_completion_callbacks = []
  @after_error_callbacks = []
end

.reset_configuration! ⇒ Configuration

Reset configuration to defaults (useful for testing).

Returns:



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

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

.running? ⇒ Boolean

Check if the processor is running.

Returns:

  • (Boolean) 


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

def running?
  !!@processor&.running?
end

.start ⇒ void

This method returns an undefined value.

Start the processor.



178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/patient_http/solid_queue.rb', line 178

def start
  @lifecycle_mutex.synchronize do
    return if @processor && !@processor.stopped?

    @processor = PatientHttp::Processor.new(configuration)
    @processor.observe(ProcessorObserver.new(@processor))
    @processor.start
  end

  register_handler
end

.stop(timeout: nil) ⇒ void

This method returns an undefined value.

Stop the processor gracefully.

Parameters:

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

    maximum time to wait for in-flight requests to complete



205
206
207
208
209
210
211
212
213
214
215
216
 
# File 'lib/patient_http/solid_queue.rb', line 205

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

  @lifecycle_mutex.synchronize do
    return unless @processor

    @processor.stop(timeout: timeout)
    @processor = nil
  end
end

.stopped? ⇒ Boolean

Check if the processor is stopped.

Returns:

  • (Boolean) 


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

def stopped?
  @processor.nil? || @processor.stopped?
end

.stopping? ⇒ Boolean

Check if the processor is stopping.

Returns:

  • (Boolean) 


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

def stopping?
  !!@processor&.stopping?
end