Class: Webhookdb::Jobs::ProcessWebhook

Inherits:
Object
  • Object
show all
Extended by:
Async::Job
Includes:
Amigo::DurableJob, Amigo::QueueBackoffJob, Amigo::SemaphoreBackoffJob
Defined in:
lib/webhookdb/jobs/process_webhook.rb

Instance Method Summary collapse

Methods included from Async::Job

extended

Methods included from Amigo::DurableJob

enabled?, ensure_jobs_tables, heartbeat, heartbeat!, included, insert_job, lock_job, poll_jobs, reconnect, replace_database_settings, set_database_setting, storage_datasets, unlock_job

Instance Method Details

#_perform(event) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 25

def _perform(event)
  self.with_log_tags(@sint.log_tags) do
    kw = event.payload[1].symbolize_keys
    svc = Webhookdb::Replicator.create(@sint)
    # kwargs contains: :headers, :body, :request_path, :request_method
    req = Webhookdb::Replicator::WebhookRequest.new(
      body: kw.fetch(:body),
      headers: kw.fetch(:headers),
      path: kw.fetch(:request_path),
      method: kw.fetch(:request_method),
    )
    svc.upsert_webhook(req)
  end
end

#before_perform(*args) ⇒ Object 



20
21
22
23
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 20

def before_perform(*args)
  event = Amigo::Event.from_json(args[0])
  @sint = self.lookup_model(Webhookdb::ServiceIntegration, event)
end

#dependent_queues ⇒ Object 



18
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 18

def dependent_queues = ["critical"]

#semaphore_expiry ⇒ Object 



17
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 17

def semaphore_expiry = 5.minutes.to_i

#semaphore_key ⇒ Object 



40
41
42
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 40

def semaphore_key
  return "semaphore-procwebhook-#{@sint.organization_id}"
end

#semaphore_size ⇒ Object 



44
45
46
 
# File 'lib/webhookdb/jobs/process_webhook.rb', line 44

def semaphore_size
  return @sint.organization.job_semaphore_size
end