Class: ExpireJob::Middleware

Inherits:
Object
  • Object
show all
Defined in:
lib/expire_job.rb

Instance Method Summary collapse

Instance Method Details

#call(worker, msg, queue, &block) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
 
# File 'lib/expire_job.rb', line 8

def call(worker, msg, queue, &block)
  if worker.respond_to?(:expire_in)
    picked_time = pick_enqueued_at(msg)
    parsed_time = parse_time(picked_time)

    if perform_expire_check(worker.expire_in, parsed_time)
      yield
    else
      logger.info { "Expired job is skipped. args=#{truncate(msg['args'].inspect)}" }
      perform_callback(worker, :after_expire, msg['args'])
      nil
    end
  else
    yield
  end
end

#logger ⇒ Object 



94
95
96
97
98
99
100
101
102
 
# File 'lib/expire_job.rb', line 94

def logger
  if defined?(::Sidekiq)
    ::Sidekiq.logger
  elsif defined?(::Rails)
    ::Rails.logger
  else
    ::Logger.new(STDOUT)
  end
end

#parse_time(value) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
 
# File 'lib/expire_job.rb', line 58

def parse_time(value)
  if value.to_s.match?(/\d+\.\d+/)
    Time.at(value)
  else
    Time.parse(value)
  end
rescue => e
  logger.warn { "Can not parse this value. value=#{value.inspect}" }
  nil
end

#perform_callback(worker, callback_name, args) ⇒ Object 



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/expire_job.rb', line 69

def perform_callback(worker, callback_name, args)
  if worker.respond_to?(callback_name)
    parameters = worker.method(callback_name).parameters

    begin
      if parameters.empty?
        worker.send(callback_name)
      else
        worker.send(callback_name, *args)
      end
    rescue ArgumentError => e
      message = "The number of parameters of the callback method (#{parameters.size}) is not the same as the number of arguments (#{args.size})"
      raise ArgumentError.new("#{self.class}:#{worker.class} #{message} callback_name=#{callback_name} args=#{args.inspect} parameters=#{parameters.inspect}")
    end
  end
end

#perform_expire_check(expire_in, enqueued_at) ⇒ Object 



25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/expire_job.rb', line 25

def perform_expire_check(expire_in, enqueued_at)
  if enqueued_at.nil?
    logger.warn { "Can not expire this job because enqueued_at is nil." }
    return true
  end

  if enqueued_at < Time.now - expire_in
    false
  else
    true
  end
end

#pick_enqueued_at(msg) ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/expire_job.rb', line 38

def pick_enqueued_at(msg)
  args = msg['args']
  enqueued_at = nil

  if args.is_a?(Array) && args.size >= 1 && args.last.is_a?(Hash)
    enqueued_at = args.last['enqueued_at']
    logger.info { "enqueued_at was found in args. enqueued_at=#{enqueued_at}" } if enqueued_at
  end

  if enqueued_at.nil?
    # The msg has both created_at and enqueued_at.
    #   created_at: is a time when #perform_async or #perform_in is called
    #   enqueued_at: is a time when the job is inserted into a queue
    enqueued_at = msg['created_at'] # TODO Use enqueued_at?
    logger.debug { "enqueued_at was found in msg. enqueued_at=#{enqueued_at}" } if enqueued_at
  end

  enqueued_at
end

#truncate(text, length: 100) ⇒ Object 



86
87
88
89
90
91
92
 
# File 'lib/expire_job.rb', line 86

def truncate(text, length: 100)
  if text.length > length
    text.slice(0, length)
  else
    text
  end
end