Module: Cosmo::Job::ClassMethods

Defined in:
lib/cosmo/job.rb,
sig/cosmo/job.rbs

Instance Method Summary collapse

Instance Method Details

#client ⇒ Client

Returns:



108
109
110
 
# File 'lib/cosmo/job.rb', line 108

def client
  @client ||= Client.instance
end

#concurrency_key(args) ⇒ ::String?

Derive the fully-scoped concurrency key for a given args array.

Parameters:

  • args (Array[untyped])

Returns:

  • (::String, nil) 


64
65
66
67
68
69
70
71
 
# File 'lib/cosmo/job.rb', line 64

def concurrency_key(args)
  config = concurrency_options
  return unless config

  base = Utils::String.underscore(name)
  suffix = config[:key]&.call(*args)
  suffix ? "#{base}/#{suffix}" : base
end

#concurrency_options ⇒ { limit: Integer, key: Proc?, duration: Integer }?

Returns a normalized concurrency config hash, or nil when not configured. Always contains :limit, :key, :duration, and :retry_in.

Returns:

  • ({ limit: Integer, key: Proc?, duration: Integer }, nil) 


50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/cosmo/job.rb', line 50

def concurrency_options
  value = default_options.dig(:limit, :concurrency)
  return unless value

  duration = default_options.dig(:limit, :duration).to_i
  retry_in = default_options.dig(:limit, :retry_in)&.to_i || (duration / 2)

  case value
  when Integer then { limit: value, key: nil, duration: duration, retry_in: retry_in }
  when Hash    then { limit: value.fetch(:to), key: value[:key], duration: duration, retry_in: retry_in }
  end
end

#default_options ⇒ Hash[Symbol, untyped]

Returns:

  • (Hash[Symbol, untyped]) 


102
103
104
 
# File 'lib/cosmo/job.rb', line 102

def default_options
  @default_options ||= (superclass.respond_to?(:default_options) ? superclass.default_options : Data::DEFAULTS).dup
end

#limits_concurrency? ⇒ Boolean

Returns:

  • (Boolean) 


44
45
46
 
# File 'lib/cosmo/job.rb', line 44

def limits_concurrency?
  !!concurrency_options
end

#options(**config) ⇒ Hash[Symbol, untyped] Also known as: cosmo_options

Parameters:

  • config (Hash) —

    a customizable set of options

Options Hash (**config):

  • :stream (Symbol) —

    NATS stream to publish to (default: :default)

  • :retry (Integer) —

    max delivery attempts before giving up (default: 3)

  • :dead (Boolean) —

    move to dead-letter stream after retries exhausted (default: true)

  • :limit (Hash) —

    execution limits:

    limit: { duration: 30 } limit: { duration: 30, concurrency: 3 } limit: { duration: 30, concurrency: { to: 3, key: ->(id) { id } } } limit: { duration: 30, concurrency: 3, retry_in: 5 }

  • :"limit[:duration]" (Integer) —

    hard execution timeout in seconds. The job thread is killed after this many seconds and counts as a failed attempt (retried with exponential backoff, moved to DLQ after retries exhausted).

  • :"limit[:concurrency]" (Integer, Hash) —

    caps how many instances run at once across all workers. Jobs that cannot acquire a slot are NAK'd (see retry_in) so they are not re-delivered until the slot is likely free. Requires duration. Pass an Integer for a class-wide cap, or { to: N, key: ->(args) {} } to scope per key.

  • :"limit[:retry_in]" (Integer) —

    seconds to wait before NATS redelivers a job that was NAK'd for lack of a concurrency slot (default: half of duration). Counts against the same delivery counter as any other retry -- a job stuck behind the concurrency limit for enough consecutive attempts is dropped/DLQ'd exactly like one that keeps failing outright.

Returns:

  • (Hash[Symbol, untyped]) 


35
36
37
38
39
40
41
 
# File 'lib/cosmo/job.rb', line 35

def options(**config)
  if config[:limit] && config.dig(:limit, :concurrency) && !config.dig(:limit, :duration).to_i.positive?
    raise ArgumentError, "limit: duration is required when concurrency is set"
  end

  default_options.merge!(config)
end

#perform(*args, async: true, **options) ⇒ ::String?

Parameters:

  • args (Object)
  • async: (Boolean) (defaults to: true)
  • options (Object)

Returns:

  • (::String, nil) 


73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/cosmo/job.rb', line 73

def perform(*args, async: true, **options)
  data = Data.new(name, args, default_options.merge(options))
  unless async
    payload = Utils::Json.parse(data.to_args[1])
    raise ArgumentError, "Cannot parse payload" unless payload

    new.perform(*payload[:args])
    return
  end

  Publisher.publish_job(data)
end

#perform_async(*args) ⇒ ::String

Parameters:

  • args (Object)

Returns:

  • (::String) 


86
87
88
 
# File 'lib/cosmo/job.rb', line 86

def perform_async(*args)
  perform(*args)
end

#perform_at(timestamp, *args) ⇒ ::String

Parameters:

  • timestamp (Integer, Time)
  • args (Object)

Returns:

  • (::String) 


90
91
92
 
# File 'lib/cosmo/job.rb', line 90

def perform_at(timestamp, *args)
  perform(*args, at: timestamp)
end

#perform_in(interval, *args) ⇒ ::String

Parameters:

  • interval (Integer, Float)
  • args (Object)

Returns:

  • (::String) 


94
95
96
 
# File 'lib/cosmo/job.rb', line 94

def perform_in(interval, *args)
  perform(*args, in: interval)
end

#perform_sync(*args) ⇒ void

This method returns an undefined value.

Parameters:

  • args (Object) 


98
99
100
 
# File 'lib/cosmo/job.rb', line 98

def perform_sync(*args)
  perform(*args, async: false)
end