Class: KubeMQ::CancellationToken

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

Overview

Note:

This class is thread-safe. All state access is synchronized via a Mutex and ConditionVariable.

Thread-safe cooperative cancellation token for stopping subscriptions.

Create a token, pass it to a subscribe_to_* method, and call #cancel to gracefully stop the subscription. Multiple threads may safely check #cancelled? or #wait concurrently.

Examples:

Cancel a subscription after 10 seconds

token = KubeMQ::CancellationToken.new
subscription = client.subscribe_to_events(sub, cancellation_token: token) do |event|
  process(event)
end
sleep 10
token.cancel
subscription.wait(5)

See Also:

Instance Method Summary collapse

Constructor Details

#initialize ⇒ CancellationToken

Creates a new uncancelled token.



25
26
27
28
29
 
# File 'lib/kubemq/cancellation_token.rb', line 25

def initialize
  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @cancelled = false
end

Instance Method Details

#cancel ⇒ void

This method returns an undefined value.

Signals cancellation and wakes all threads waiting on this token.

This method is idempotent — calling it multiple times is safe.



36
37
38
39
40
41
 
# File 'lib/kubemq/cancellation_token.rb', line 36

def cancel
  @mutex.synchronize do
    @cancelled = true
    @condition.broadcast
  end
end

#cancelled? ⇒ Boolean

Returns whether cancellation has been signalled.

Returns:

  • (Boolean) —

    true if #cancel has been called



46
47
48
 
# File 'lib/kubemq/cancellation_token.rb', line 46

def cancelled?
  @mutex.synchronize { @cancelled }
end

#wait(timeout: nil) ⇒ Boolean

Blocks the calling thread until the token is cancelled or the timeout elapses.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum seconds to wait (+nil+ for indefinite)

Returns:

  • (Boolean) —

    true if cancelled, false if timed out



54
55
56
57
58
59
60
61
 
# File 'lib/kubemq/cancellation_token.rb', line 54

def wait(timeout: nil)
  @mutex.synchronize do
    return true if @cancelled

    @condition.wait(@mutex, timeout)
    @cancelled
  end
end