Class: JobWorkflow::Semaphore

Inherits:
Object
  • Object
show all
Defined in:
lib/job_workflow/semaphore.rb,
sig/generated/job_workflow/semaphore.rbs

Constant Summary collapse

DEFAULT_POLLING_INTERVAL =

Signature:

  • Float

Returns:

  • (Float) 
3.0

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(concurrency_key:, concurrency_duration:, concurrency_limit: 1, polling_interval: DEFAULT_POLLING_INTERVAL) ⇒ Semaphore

: ( concurrency_key: String, concurrency_duration: ActiveSupport::Duration, ?concurrency_limit: Integer, ?polling_interval: Float ) -> void

Parameters:

  • concurrency_key: (String)
  • concurrency_duration: (ActiveSupport::Duration)
  • concurrency_limit: (Integer) (defaults to: 1)
  • polling_interval: (Float) (defaults to: DEFAULT_POLLING_INTERVAL) 


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

def initialize(
  concurrency_key:,
  concurrency_duration:,
  concurrency_limit: 1,
  polling_interval: DEFAULT_POLLING_INTERVAL
)
  @concurrency_key = concurrency_key
  @concurrency_duration = concurrency_duration
  @concurrency_limit = concurrency_limit
  @polling_interval = polling_interval
end

Instance Attribute Details

#concurrency_duration ⇒ ActiveSupport::Duration (readonly)

Signature:

  • ActiveSupport::Duration

Returns:

  • (ActiveSupport::Duration) 


10
11
12
 
# File 'lib/job_workflow/semaphore.rb', line 10

def concurrency_duration
  @concurrency_duration
end

#concurrency_key ⇒ String (readonly)

Signature:

  • String

Returns:

  • (String) 


8
9
10
 
# File 'lib/job_workflow/semaphore.rb', line 8

def concurrency_key
  @concurrency_key
end

#concurrency_limit ⇒ Integer (readonly)

Signature:

  • Integer

Returns:

  • (Integer) 


9
10
11
 
# File 'lib/job_workflow/semaphore.rb', line 9

def concurrency_limit
  @concurrency_limit
end

#polling_interval ⇒ Float (readonly)

Signature:

  • Float

Returns:

  • (Float) 


69
70
71
 
# File 'lib/job_workflow/semaphore.rb', line 69

def polling_interval
  @polling_interval
end

Class Method Details

.available? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


14
15
16
 
# File 'lib/job_workflow/semaphore.rb', line 14

def available?
  QueueAdapter.current.semaphore_available?
end

Instance Method Details

#signal ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


51
52
53
54
55
56
57
 
# File 'lib/job_workflow/semaphore.rb', line 51

def signal
  return true unless self.class.available?

  result = QueueAdapter.current.semaphore_signal(self)
  Instrumentation.notify_throttle_release(self)
  result
end

#wait ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/job_workflow/semaphore.rb', line 38

def wait
  return true unless self.class.available?

  Instrumentation.instrument_throttle(self) do
    loop do
      return true if QueueAdapter.current.semaphore_wait(self)

      sleep(polling_interval)
    end
  end
end

#with ⇒ void

This method returns an undefined value.

: [T] () { () -> T } -> T



60
61
62
63
64
65
 
# File 'lib/job_workflow/semaphore.rb', line 60

def with(&)
  wait
  yield
ensure
  signal
end