Class: PgEventstore::SubscriptionsManager

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/pg_eventstore/subscriptions/subscriptions_manager.rb,
sig/pg_eventstore/subscriptions/subscriptions_manager.rbs

Overview

The public Subscriptions API, available to the user.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config:, set_name:, max_retries: nil, retries_interval: nil, force_lock: false) ⇒ SubscriptionsManager

Returns a new instance of SubscriptionsManager.

Parameters:

  • config (PgEventstore::Config)
  • set_name (String)
  • max_retries (Integer, nil) (defaults to: nil) —

    max number of retries of failed SubscriptionsSet

  • retries_interval (Integer, nil) (defaults to: nil) —

    a delay between retries of failed SubscriptionsSet

  • force_lock (Boolean) (defaults to: false) —

    whether to force-lock subscriptions



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 55

def initialize(config:, set_name:, max_retries: nil, retries_interval: nil, force_lock: false)
  @config = config
  @set_name = set_name
  @subscriptions_set_lifecycle = SubscriptionsSetLifecycle.new(
    config_name,
    {
      name: set_name,
      max_restarts_number: max_retries || config.subscriptions_set_max_retries,
      time_between_restarts: retries_interval || config.subscriptions_set_retries_interval,
    }
  )
  @subscriptions_lifecycle = SubscriptionsLifecycle.new(
    config_name, @subscriptions_set_lifecycle, force_lock:
  )
  @subscription_feeder = SubscriptionFeeder.new(
    config_name:,
    subscriptions_set_lifecycle: @subscriptions_set_lifecycle,
    subscriptions_lifecycle: @subscriptions_lifecycle
  )
end

Instance Attribute Details

#config ⇒ Config

Returns the value of attribute config.

Returns:



45
46
47
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 45

def config
  @config
end

Class Method Details

.callbacks ⇒ Callbacks

Returns:



10
 
# File 'sig/pg_eventstore/subscriptions/subscriptions_manager.rbs', line 10

def self.callbacks: () -> Callbacks

Instance Method Details

#config_name ⇒ Symbol

Returns:

  • (Symbol) 


211
212
213
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 211

def config_name
  @config.name
end

#connection ⇒ PgEventstore::Connection

Returns:



252
253
254
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 252

def connection
  PgEventstore.connection(config_name)
end

#create_replication(subscription_name, replica_config_name, options: {}, max_events_to_replicate: config.max_events_to_replicate, pull_interval: config.subscription_pull_interval, max_retries: config.subscription_max_retries, retries_interval: config.subscription_retries_interval, restart_terminator: config.subscription_restart_terminator, failed_subscription_notifier: config.failed_subscription_notifier, graceful_shutdown_timeout: config.subscription_graceful_shutdown_timeout) ⇒ void

This method returns an undefined value.

Parameters:

  • subscription_name (String) —

    subscription's name. For example, you can name it after the target database: "Target: my_eventstore_replica1"

  • replica_config_name (Symbol) —

    the destination replica database config to replicate events into. You have to have it preconfigured with PgEventstore.configure(). Example: PgEventstore.configure(name: :my_eventstore_replica1) do |config| config.pg_uri = 'postgresql://postgres:postgres@localhost:5432/my_eventstore_replica1' end Refer to the docs/configuration.md for more info

  • options (Hash) (defaults to: {}) —

    subscription options

  • max_events_to_replicate (Integer) (defaults to: config.max_events_to_replicate) —

    set the upper limit of how many events the handler can copy per turn

  • pull_interval (Integer, Float) (defaults to: config.subscription_pull_interval) —

    an interval in seconds to determine how often to query new events of the given subscription.

  • max_retries (Integer) (defaults to: config.subscription_max_retries) —

    max number of retries of failed Subscription

  • retries_interval (Integer, Float) (defaults to: config.subscription_retries_interval) —

    a delay between retries of failed Subscription

  • restart_terminator (#call, nil) (defaults to: config.subscription_restart_terminator) —

    a callable object which is invoked with PgEventstore::Subscription instance to determine whether restarts should be stopped(true - stops restarts, false - continues restarts)

  • failed_subscription_notifier (#call, nil) (defaults to: config.failed_subscription_notifier) —

    a callable object which is invoked with PgEventstore::Subscription instance and error instance after the related subscription died due to error and no longer can be automatically restarted due to max retries number reached. You can use this hook to send a notification about failed subscription.

  • graceful_shutdown_timeout (Integer, Float) (defaults to: config.subscription_graceful_shutdown_timeout) —

    the number of seconds to wait until force-shutdown the subscription during the stop process

Options Hash (options:):

  • :from_position (Integer) —

    a starting subscription position

  • :filter (Hash) —

    provide it to filter events. It works the same way as a :filter option of Client#read method. Use this option when you want to replicate certain set of events.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 153

def create_replication(subscription_name,
                       replica_config_name,
                       options: {},
                       max_events_to_replicate: config.max_events_to_replicate,
                       pull_interval: config.subscription_pull_interval,
                       max_retries: config.subscription_max_retries,
                       retries_interval: config.subscription_retries_interval,
                       restart_terminator: config.subscription_restart_terminator,
                       failed_subscription_notifier: config.failed_subscription_notifier,
                       graceful_shutdown_timeout: config.subscription_graceful_shutdown_timeout)
  Utils.assert_node_role!(config, [Config::NodeRole::PRIMARY])
  subscription = Subscription.using_connection(config.name).new(
    set: @set_name, name: subscription_name, options:, chunk_query_interval: pull_interval,
    max_restarts_number: max_retries, time_between_restarts: retries_interval
  )
  runner = ReplicaSubscriptionRunner.new(
    stats: SubscriptionHandlerPerformance.new,
    events_processor: EventsProcessor.new(
      graceful_shutdown_timeout:,
      consumer: EventsProcessorConsumer::Replica.create_consumer(config_name, replica_config_name),
      recovery_strategies: recovery_strategies(subscription, restart_terminator, failed_subscription_notifier)
    ),
    subscription:,
    max_events_per_chunk: max_events_to_replicate,
    initial_events_per_chunk: max_events_to_replicate
  )

  @subscriptions_lifecycle.runners.push(runner)
  true
end

#deserializer(middlewares) ⇒ EventDeserializer

Parameters:

  • middlewares (::Array[Symbol], nil)

Returns:

  • (EventDeserializer) 


224
225
226
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 224

def deserializer(middlewares)
  EventDeserializer.new(select_middlewares(middlewares), config.event_class_resolver)
end

#recovery_strategies(subscription, restart_terminator, failed_subscription_notifier) ⇒ Array<PgEventstore::RunnerRecoveryStrategy>

Parameters:

Returns:



240
241
242
243
244
245
246
247
248
249
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 240

def recovery_strategies(subscription, restart_terminator, failed_subscription_notifier)
  [
    RunnerRecoveryStrategies::RestoreConnection.new(config_name),
    RunnerRecoveryStrategies::RestoreSubscriptionRunner.new(
      subscription:,
      restart_terminator:,
      failed_subscription_notifier:
    ),
  ]
end

#run_cli_callbacks { ... } ⇒ Object

Returns the result of the passed block.

Yields:

Yield Returns:

  • (Object)

Returns:

  • (Object) —

    the result of the passed block



218
219
220
221
222
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 218

def run_cli_callbacks(&)
  return yield unless defined?(::PgEventstore::CLI)

  PgEventstore::CLI.callbacks.run_callbacks(:start_manager, self, &)
end

#select_middlewares(middlewares = nil) ⇒ Array<PgEventstore::Middleware>

Parameters:

  • middlewares (Array, nil) (defaults to: nil)

Returns:



230
231
232
233
234
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 230

def select_middlewares(middlewares = nil)
  return config.middlewares.values unless middlewares

  config.middlewares.slice(*middlewares).values
end

#start ⇒ PgEventstore::BasicRunner?

Returns:



203
204
205
206
207
208
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 203

def start
  start!
rescue PgEventstore::SubscriptionAlreadyLockedError => e
  PgEventstore.logger&.warn(e.message)
  nil
end

#start! ⇒ PgEventstore::BasicRunner

Returns:

Raises:



196
197
198
199
200
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 196

def start!
  run_cli_callbacks do
    @subscription_feeder.start
  end
end

#subscribe(subscription_name, handler:, options: {}, middlewares: nil, pull_interval: config.subscription_pull_interval, max_retries: config.subscription_max_retries, retries_interval: config.subscription_retries_interval, restart_terminator: config.subscription_restart_terminator, failed_subscription_notifier: config.failed_subscription_notifier, graceful_shutdown_timeout: config.subscription_graceful_shutdown_timeout, in_batches: false) ⇒ void

This method returns an undefined value.

Parameters:

  • subscription_name (String) —

    subscription's name

  • handler (#call) —

    subscription's handler

  • options (Hash) (defaults to: {}) —

    subscription options

  • middlewares (Array<Symbol>, nil) (defaults to: nil) —

    provide a list of middleware names to override a config's middlewares

  • pull_interval (Integer, Float) (defaults to: config.subscription_pull_interval) —

    an interval in seconds to determine how often to query new events of the given subscription.

  • max_retries (Integer) (defaults to: config.subscription_max_retries) —

    max number of retries of failed Subscription

  • retries_interval (Integer, Float) (defaults to: config.subscription_retries_interval) —

    a delay between retries of failed Subscription

  • restart_terminator (#call, nil) (defaults to: config.subscription_restart_terminator) —

    a callable object which is invoked with PgEventstore::Subscription instance to determine whether restarts should be stopped(true - stops restarts, false - continues restarts)

  • failed_subscription_notifier (#call, nil) (defaults to: config.failed_subscription_notifier) —

    a callable object which is invoked with PgEventstore::Subscription instance and error instance after the related subscription died due to error and no longer can be automatically restarted due to max retries number reached. You can use this hook to send a notification about failed subscription.

  • graceful_shutdown_timeout (integer, Float) (defaults to: config.subscription_graceful_shutdown_timeout) —

    the number of seconds to wait until force-shutdown the subscription during the stop process

  • in_batches (Boolean) (defaults to: false) —

    whether a batch of events should be yielded instead a single event

Options Hash (options:):

  • :from_position (Integer) —

    a starting subscription position

  • :resolve_link_tos (Boolean) —

    When using projections to create new events you can set whether the generated events are pointers to existing events. Setting this option to true tells PgEventstore to return the original event instead a link event.

  • :filter (Hash) —

    provide it to filter events. It works the same way as a :filter option of Client#read method. Filtering by both - event types and streams are available.



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 100

def subscribe(subscription_name, handler:, options: {}, middlewares: nil,
              pull_interval: config.subscription_pull_interval,
              max_retries: config.subscription_max_retries,
              retries_interval: config.subscription_retries_interval,
              restart_terminator: config.subscription_restart_terminator,
              failed_subscription_notifier: config.failed_subscription_notifier,
              graceful_shutdown_timeout: config.subscription_graceful_shutdown_timeout,
              in_batches: false)
  subscription = Subscription.using_connection(config.name).new(
    set: @set_name, name: subscription_name, options:, chunk_query_interval: pull_interval,
    max_restarts_number: max_retries, time_between_restarts: retries_interval
  )
  consumer_class = EventsProcessorConsumer.consumer_class(in_batches)
  runner = SubscriptionRunner.new(
    stats: SubscriptionHandlerPerformance.new,
    events_processor: EventsProcessor.new(
      graceful_shutdown_timeout:,
      consumer: consumer_class.create_consumer(handler, deserializer(middlewares)),
      recovery_strategies: recovery_strategies(subscription, restart_terminator, failed_subscription_notifier)
    ),
    subscription:
  )

  @subscriptions_lifecycle.runners.push(runner)
  true
end

#subscriptions ⇒ Array<PgEventstore::Subscription>

Returns:



185
186
187
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 185

def subscriptions
  @subscriptions_lifecycle.subscriptions.map(&:dup)
end

#subscriptions_set ⇒ PgEventstore::SubscriptionsSet?

Returns:



190
191
192
 
# File 'lib/pg_eventstore/subscriptions/subscriptions_manager.rb', line 190

def subscriptions_set
  @subscriptions_set_lifecycle.subscriptions_set&.dup
end