Class: Wurk::Fetcher::Reliable::UnitOfWork

Inherits:
Struct
  • Object
show all
Defined in:
lib/wurk/fetcher/reliable.rb

Overview

Carries the public queue key, the raw (still-JSON) job payload, and the capsule we use to reach Redis. ACK removes from the private list; requeue pushes back to the public queue head so the job is next pulled. LREM count=1 is idempotent for our payloads since each job's JSON contains a unique jid.

jid is filled in by the Processor once it has parsed the payload — the fetcher never parses. It is only used to retire the job's poison-pill recovery counter, so an ACK without one is still a complete ACK.

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#config ⇒ Object

Returns the value of attribute config

Returns:

  • (Object) —

    the current value of config



51
52
53
 
# File 'lib/wurk/fetcher/reliable.rb', line 51

def config
  @config
end

#jid ⇒ Object

Returns the value of attribute jid

Returns:

  • (Object) —

    the current value of jid



51
52
53
 
# File 'lib/wurk/fetcher/reliable.rb', line 51

def jid
  @jid
end

#job ⇒ Object

Returns the value of attribute job

Returns:

  • (Object) —

    the current value of job



51
52
53
 
# File 'lib/wurk/fetcher/reliable.rb', line 51

def job
  @job
end

#queue ⇒ Object

Returns the value of attribute queue

Returns:

  • (Object) —

    the current value of queue



51
52
53
 
# File 'lib/wurk/fetcher/reliable.rb', line 51

def queue
  @queue
end

Instance Method Details

#acknowledge ⇒ Object

The counter DEL rides this round trip rather than taking one of its own: the ACK is the only Redis call the success path makes, and a per-job call would be a fetch+execute regression for the sake of a key that exists for roughly no jobs. See Middleware::PoisonPill.



56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/wurk/fetcher/reliable.rb', line 56

def acknowledge
  private_list = Reliable.private_queue_name(queue)
  job_jid = jid.to_s
  return config.redis { |conn| conn.call('LREM', private_list, 1, job) } if job_jid.empty?

  config.redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('LREM', private_list, 1, job)
      Middleware::PoisonPill.clear_in(pipe, job_jid)
    end
  end
end

#queue_name ⇒ Object 



69
70
71
 
# File 'lib/wurk/fetcher/reliable.rb', line 69

def queue_name
  queue.delete_prefix(Keys::QUEUE_PREFIX)
end

#requeue ⇒ Object 



73
74
75
 
# File 'lib/wurk/fetcher/reliable.rb', line 73

def requeue
  config.redis { |conn| conn.call('RPUSH', queue, job) }
end