Class: KubeMQ::Subscription

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

Overview

Note:

This class is thread-safe. Status transitions and error state are protected by a mutex.

Handle for an active subscription running on a background thread.

Returned by subscribe_to_* methods on PubSubClient and CQClient. Use #cancel to stop the subscription, #active? to check its state, and #wait or #join to block until the subscription thread exits.

Examples:

Cancel a subscription gracefully

subscription = client.subscribe_to_events(sub) { |e| process(e) }
# ... later ...
subscription.cancel
subscription.wait(5) # wait up to 5 seconds for thread to exit

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(thread:, cancellation_token:) ⇒ Subscription

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Creates a new subscription handle.

Parameters:

  • thread (Thread) —

    the background thread running the subscription loop

  • cancellation_token (CancellationToken) —

    token used to signal cancellation



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

def initialize(thread:, cancellation_token:)
  @thread = thread
  @cancellation_token = cancellation_token
  @status = :active
  @last_error = nil
  @mutex = Mutex.new
end

Instance Attribute Details

#last_error ⇒ Object (readonly)

Returns the value of attribute last_error.



28
 
# File 'lib/kubemq/subscription.rb', line 28

attr_reader :status, :last_error

#status ⇒ Symbol (readonly)

Returns current subscription state — :active, :cancelled, :error, or :closed.

Returns:

  • (Symbol) —

    current subscription state — :active, :cancelled, :error, or :closed



28
29
30
 
# File 'lib/kubemq/subscription.rb', line 28

def status
  @status
end

Instance Method Details

#active? ⇒ Boolean

Returns whether the subscription is still actively receiving messages.

Returns:

  • (Boolean) —

    true if status is :active and the thread is alive



57
58
59
 
# File 'lib/kubemq/subscription.rb', line 57

def active?
  @mutex.synchronize { @status == :active } && @thread.alive?
end

#cancel ⇒ void

This method returns an undefined value.

Cancels the subscription, stops the gRPC stream, and waits up to 5 seconds for the background thread to exit.



47
48
49
50
51
52
 
# File 'lib/kubemq/subscription.rb', line 47

def cancel
  @mutex.synchronize { @status = :cancelled }
  @cancellation_token.cancel
  begin; @thread[:grpc_call]&.cancel; rescue StandardError; end
  @thread.join(5)
end

#join(timeout = nil) ⇒ Thread?

Alias for #wait. Blocks until the subscription thread exits.

Parameters:

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

    maximum seconds to wait (+nil+ for indefinite)

Returns:

  • (Thread, nil) —

    the subscription thread if it exited, nil on timeout



74
75
76
 
# File 'lib/kubemq/subscription.rb', line 74

def join(timeout = nil)
  @thread.join(timeout)
end

#mark_closed ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Marks this subscription as closed.



94
95
96
 
# File 'lib/kubemq/subscription.rb', line 94

def mark_closed
  @mutex.synchronize { @status = :closed }
end

#mark_error(error) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Records an error on this subscription.

Parameters:

  • error (StandardError) —

    the error that occurred



83
84
85
86
87
88
 
# File 'lib/kubemq/subscription.rb', line 83

def mark_error(error)
  @mutex.synchronize do
    @status = :error
    @last_error = error
  end
end

#wait(timeout = nil) ⇒ Thread?

Blocks the calling thread until the subscription thread exits or the timeout elapses.

Parameters:

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

    maximum seconds to wait (+nil+ for indefinite)

Returns:

  • (Thread, nil) —

    the subscription thread if it exited, nil on timeout



66
67
68
 
# File 'lib/kubemq/subscription.rb', line 66

def wait(timeout = nil)
  @thread.join(timeout)
end