Class: KubeMQ::PubSub::EventsStoreSubscription

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

Overview

Configuration for subscribing to durable events store channels.

Unlike EventsSubscription, events store subscriptions support replay from a historical position. Pass to KubeMQ::PubSubClient#subscribe_to_events_store with a block to receive EventStoreReceived messages.

Examples:

Subscribe from the beginning

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders.created",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
)
client.subscribe_to_events_store(sub) { |event| puts event.body }

Subscribe from a specific sequence

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders.created",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_SEQUENCE,
  start_position_value: 42
)

See Also:

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(channel:, start_position:, start_position_value: 0, group: nil) ⇒ EventsStoreSubscription

Returns a new instance of EventsStoreSubscription.

Parameters:

  • channel (String) —

    channel name (required)

  • start_position (Integer) —

    one of the KubeMQ::PubSub::EventStoreStartPosition constants (required)

  • start_position_value (Integer) (defaults to: 0) —

    context value for the chosen start position (default: 0)

  • group (String, nil) (defaults to: nil) —

    consumer group name (default: nil)



65
66
67
68
69
70
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 65

def initialize(channel:, start_position:, start_position_value: 0, group: nil)
  @channel = channel
  @start_position = start_position
  @start_position_value = start_position_value
  @group = group
end

Instance Attribute Details

#channel ⇒ String

Returns channel name to subscribe to.

Returns:

  • (String) —

    channel name to subscribe to



59
60
61
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 59

def channel
  @channel
end

#group ⇒ String?

Returns consumer group for load-balanced delivery.

Returns:

  • (String, nil) —

    consumer group for load-balanced delivery



59
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 59

attr_accessor :channel, :group, :start_position, :start_position_value

#start_position ⇒ Integer

Returns one of the KubeMQ::PubSub::EventStoreStartPosition constants.

Returns:



59
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 59

attr_accessor :channel, :group, :start_position, :start_position_value

#start_position_value ⇒ Integer

Returns sequence number, Unix timestamp, or delta depending on start_position.

Returns:

  • (Integer) —

    sequence number, Unix timestamp, or delta depending on start_position



59
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 59

attr_accessor :channel, :group, :start_position, :start_position_value

Instance Method Details

#subscribe_type ⇒ Integer

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.

Returns the subscribe type constant for the transport layer.

Returns:



76
77
78
 
# File 'lib/kubemq/pubsub/events_store_subscription.rb', line 76

def subscribe_type
  SubscribeType::EVENTS_STORE
end