Module: PatientHttp::Sidekiq

Defined in:
lib/patient_http/sidekiq.rb,
lib/patient_http/sidekiq/stats.rb,
lib/patient_http/sidekiq/web_ui.rb,
lib/patient_http/sidekiq/context.rb,
lib/patient_http/sidekiq/task_handler.rb,
lib/patient_http/sidekiq/task_monitor.rb,
lib/patient_http/sidekiq/configuration.rb,
lib/patient_http/sidekiq/request_worker.rb,
lib/patient_http/sidekiq/callback_worker.rb,
lib/patient_http/sidekiq/lifecycle_hooks.rb,
lib/patient_http/sidekiq/request_executor.rb,
lib/patient_http/sidekiq/processor_observer.rb,
lib/patient_http/sidekiq/task_monitor_thread.rb

Defined Under Namespace

Classes: CallbackWorker, Configuration, Context, LifecycleHooks, ProcessorObserver, RequestExecutor, RequestWorker, Stats, TaskHandler, TaskMonitor, TaskMonitorThread, WebUI

Constant Summary collapse

VERSION = 
File.read(File.expand_path("../../../VERSION", __FILE__)).strip

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.configuration ⇒ Configuration

Ensure configuration is initialized

Returns:



120
121
122
 
# File 'lib/patient_http/sidekiq.rb', line 120

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 (internal accessor)

Returns:

  • (PatientHttp::Processor, nil) 


388
389
390
 
# File 'lib/patient_http/sidekiq.rb', line 388

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



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

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 that was raised



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

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

.append_middleware ⇒ void

This method returns an undefined value.

Add Sidekiq middleware for context handling. The middleware is already added during initialization. You can call this method again to append the middleware if needed to insert it after other middleware. If you need further control, you can manually add the PatientHttp::Sidekiq::Context::Middleware middleware yourself.



155
156
157
158
159
160
161
 
# File 'lib/patient_http/sidekiq.rb', line 155

def append_middleware
  ::Sidekiq.configure_server do |config|
    config.server_middleware do |chain|
      chain.add PatientHttp::Sidekiq::Context::Middleware
    end
  end
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:



108
109
110
111
112
113
114
115
116
 
# File 'lib/patient_http/sidekiq.rb', line 108

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

.decrypt(data) ⇒ Hash

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.

Decrypt data using the configured encryptor.

Parameters:

  • data (Hash) —

    the data to decrypt

Returns:

  • (Hash) —

    the decrypted data (or original if not encrypted)



214
215
216
 
# File 'lib/patient_http/sidekiq.rb', line 214

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

.draining? ⇒ Boolean

Check if the processor is draining (not accepting new requests but still processing in-flight ones).

Returns:

  • (Boolean) 


174
175
176
 
# File 'lib/patient_http/sidekiq.rb', line 174

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

.encrypt(data) ⇒ Hash

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.

Encrypt data using the configured encryptor.

Parameters:

  • data (Hash) —

    the data to encrypt

Returns:

  • (Hash) —

    the encrypted data (or original if no encryption configured)



205
206
207
 
# File 'lib/patient_http/sidekiq.rb', line 205

def encrypt(data)
  configuration.encryptor.encrypt(data)
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 via the PatientHttp::Response/PatientHttp::Error object. Must respond to to_h and contain only JSON-native types (nil, true, false, String, Integer, Float, Array, Hash). All hash keys will be converted to strings for serialization. Access via response.callback_args or error.callback_args using symbol or string keys.

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

    If true, treats non-2xx responses as errors and calls on_error instead of on_complete. Defaults to false.

Returns:

  • (String) —

    the request ID



257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/patient_http/sidekiq.rb', line 257

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

  options = current_sidekiq_options
  if options&.any?
    queue = options["queue"]
    options = options.merge("patient_http_callback_queue" => queue.to_s) if queue
    RequestWorker.set(options).perform_async(data, callback_name, raise_error_responses, callback_args, request_id)
  else
    RequestWorker.perform_async(data, callback_name, raise_error_responses, callback_args, request_id)
  end

  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) 


196
197
198
 
# File 'lib/patient_http/sidekiq.rb', line 196

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



367
368
369
370
371
 
# File 'lib/patient_http/sidekiq.rb', line 367

def invoke_completion_callbacks(response)
  @after_completion_callbacks.each do |callback|
    callback.call(response)
  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 that was raised



378
379
380
381
382
 
# File 'lib/patient_http/sidekiq.rb', line 378

def invoke_error_callbacks(error)
  @after_error_callbacks.each do |callback|
    callback.call(error)
  end
end

.quiet ⇒ void

This method returns an undefined value.

Signal the processor to drain (stop accepting new requests)



321
322
323
324
325
326
327
 
# File 'lib/patient_http/sidekiq.rb', line 321

def quiet
  @lifecycle_mutex.synchronize do
    return unless running?

    @processor.drain
  end
end

.register_handler ⇒ void

This method returns an undefined value.

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



287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/patient_http/sidekiq.rb', line 287

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)



350
351
352
353
354
355
356
357
358
359
360
 
# File 'lib/patient_http/sidekiq.rb', line 350

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

.reset_configuration! ⇒ Configuration

Reset configuration to defaults (useful for testing)

Returns:



126
127
128
129
130
 
# File 'lib/patient_http/sidekiq.rb', line 126

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

.running? ⇒ Boolean

Check if the processor is running.

Returns:

  • (Boolean) 


166
167
168
 
# File 'lib/patient_http/sidekiq.rb', line 166

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

.start ⇒ void

This method returns an undefined value.

Start the processor



303
304
305
306
307
308
309
310
311
312
313
314
315
316
 
# File 'lib/patient_http/sidekiq.rb', line 303

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

    @processor = PatientHttp::Processor.new(configuration)
    @processor.observe(ProcessorObserver.new(@processor))
    configuration.observers.each do |observer|
      @processor.observe(observer)
    end
    @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



333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/patient_http/sidekiq.rb', line 333

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 or has not been started.

Returns:

  • (Boolean) 


188
189
190
 
# File 'lib/patient_http/sidekiq.rb', line 188

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

.stopping? ⇒ Boolean

Check if the processor is in the process of stopping.

Returns:

  • (Boolean) 


181
182
183
 
# File 'lib/patient_http/sidekiq.rb', line 181

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

.with_sidekiq_options(options) { ... } ⇒ Object

Set Sidekiq job options for HTTP requests enqueued within the block. Options are applied with Sidekiq's set method (queue, retry, etc.). Nested calls merge options with the innermost values taking precedence. If the options include a queue, the callback job for the request is enqueued on that queue as well. Options only apply to requests enqueued in the same fiber as the block. This method has no effect when jobs run inline with Sidekiq::Testing.inline!.

Parameters:

  • options (Hash) —

    Sidekiq job options (symbol or string keys)

Yields:

  • block within which enqueued requests use the options

Returns:

  • (Object) —

    the return value of the block

Raises:

  • (ArgumentError) 


229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/patient_http/sidekiq.rb', line 229

def with_sidekiq_options(options)
  unless options.is_a?(Hash)
    raise ArgumentError.new("options must be a Hash, got: #{options.class}")
  end
  raise ArgumentError.new("with_sidekiq_options requires a block") unless block_given?

  previous = Thread.current[:patient_http_sidekiq_options]
  begin
    Thread.current[:patient_http_sidekiq_options] = (previous || {}).merge(options.transform_keys(&:to_s))
    yield
  ensure
    Thread.current[:patient_http_sidekiq_options] = previous
  end
end