Class: PgEventstore::Subscription

Inherits:

Object

Object
PgEventstore::Subscription

show all

Includes:: Extensions::OptionsExtension

Defined in:: lib/pg_eventstore/subscriptions/subscription.rb,
sig/pg_eventstore/subscriptions/subscription.rbs

Overview

Defines ruby's representation of subscriptions record.

Constant Summary collapse

MIN_EVENTS_PULL_INTERVAL = Determines the minimal allowed value of events pull frequency of the particular subscription. You can find similar constant - SubscriptionFeeder::EVENTS_PULL_INTERVAL. Unlike it - this one is responsible to detect whether the subscription should be included in the subscriptions list to query next chunk of events. Thus, this setting only determines whether it is time to make a request, but how frequent would be the actual request - determines SubscriptionFeeder::EVENTS_PULL_INTERVAL. See Also: PgEventstore::SubscriptionFeeder::EVENTS_PULL_INTERVAL

0.2

DEFAULT_TIMESTAMP =

Time.at(0).utc.freeze

Instance Attribute Summary collapse

#average_event_processing_time ⇒ Float^?
A speed of the subscription.
#chunk_query_interval ⇒ Integer, ...
Determines how often to pull events for the given Subscription in seconds.
#created_at ⇒ Time^?
#current_position ⇒ Integer^?
Current Subscription's position.
#id ⇒ Integer^?
#last_chunk_fed_at ⇒ Time^?
Shows the time when last time events were fed to the event's processor.
#last_chunk_greatest_position ⇒ Integer^?
Shows the greatest global_position of the last event in the last chunk fed to the event's processor.
#last_error ⇒ Hash^?
The information about last error caused when processing events by the Subscription.
#last_error_occurred_at ⇒ Time^?
The time when the last error occurred.
#last_restarted_at ⇒ Time^?
Last time the Subscription was restarted.
#locked_by ⇒ Integer^?
The id of subscription manager which obtained the lock of the Subscription.
#max_restarts_number ⇒ Integer^?
Maximum number of times the Subscription can be restarted.
#name ⇒ String^?
Subscription's name.
#options ⇒ Hash^?
Subscription's options to be used to query events.
#restart_count ⇒ Integer^?
The number of Subscription's restarts after its failure.
#set ⇒ String^?
Subscription's set.
#state ⇒ String^?
Current Subscription's state.
#time_between_restarts ⇒ Integer^?
Interval in seconds between retries of failed Subscription.
#total_processed_events ⇒ Integer^?
Total number of events, processed by this subscription.
#updated_at ⇒ Time^?

Class Method Summary collapse

.ping_all(subscriptions_set_id, subscriptions) ⇒ void
@param subscriptions_set_id — SubscriptionsSet#id.
.subscription_queries ⇒ PgEventstore::SubscriptionQueries

Instance Method Summary collapse

#==(other) ⇒ Boolean
@param another.
#assign_attributes(attrs) ⇒ ::Hash[untyped, untyped]
@param attrs.
#dup ⇒ PgEventstore::Subscription
Dup the current object without assigned connection.
#eql?(other) ⇒ Boolean
@param another.
#hash ⇒ Integer
#init_default_values ⇒ void
@param options.
#initialize ⇒ Subscription constructor
A new instance of Subscription.
#lock!(lock_id, force: false) ⇒ PgEventstore::Subscription
@param lock_id.
#options_hash ⇒ ::Hash[untyped, untyped]
#readonly! ⇒ Boolean
@param opt_name.
#readonly? ⇒ Boolean
@param opt_name.
#readonly_error ⇒ void
@param opt_name.
#reload ⇒ PgEventstore::Subscription
#reset_runtime_attributes ⇒ void
#subscription_queries ⇒ PgEventstore::SubscriptionQueries
#update(attrs) ⇒ ::Hash[untyped, untyped]
@param attrs.

Methods included from Extensions::OptionsExtension

included

Constructor Details

#initialize ⇒ `Subscription`

44	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 44 def initialize: (**untyped options) -> void

Instance Attribute Details

#average_event_processing_time ⇒ `Float`^?

49	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 49 attribute(:average_event_processing_time)

#chunk_query_interval ⇒ `Integer`, ...

71	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 71 attribute(:chunk_query_interval)

#created_at ⇒ `Time`^?

85	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 85 attribute(:created_at)

#current_position ⇒ `Integer`^?

41	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 41 attribute(:current_position)

#id ⇒ `Integer`^?

24	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 24 attribute(:id)

#last_chunk_fed_at ⇒ `Time`^?

74	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 74 attribute(:last_chunk_fed_at)

#last_chunk_greatest_position ⇒ `Integer`^?

78	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 78 attribute(:last_chunk_greatest_position)

#last_error ⇒ `Hash`^?

65	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 65 attribute(:last_error)

#last_error_occurred_at ⇒ `Time`^?

68	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 68 attribute(:last_error_occurred_at)

#last_restarted_at ⇒ `Time`^?

61	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 61 attribute(:last_restarted_at)

#locked_by ⇒ `Integer`^?

82	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 82 attribute(:locked_by)

#max_restarts_number ⇒ `Integer`^?

55	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 55 attribute(:max_restarts_number)

#name ⇒ `String`^?

30	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 30 attribute(:name)

#options ⇒ `Hash`^?

37	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 37 attribute(:options)

#restart_count ⇒ `Integer`^?

52	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 52 attribute(:restart_count)

#set ⇒ `String`^?

27	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 27 attribute(:set)

#state ⇒ `String`^?

45	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 45 attribute(:state)

#time_between_restarts ⇒ `Integer`^?

58	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 58 attribute(:time_between_restarts)

#total_processed_events ⇒ `Integer`^?

33	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 33 attribute(:total_processed_events)

#updated_at ⇒ `Time`^?

88	# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 88 attribute(:updated_at)

Class Method Details

.ping_all(subscriptions_set_id, subscriptions) ⇒ `void`

This method returns an undefined value.

@param subscriptions_set_id — SubscriptionsSet#id

@param subscriptions

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 94

def ping_all(subscriptions_set_id, subscriptions)
  result = subscription_queries.ping_all(subscriptions_set_id, subscriptions.map(&:id))
  subscriptions.each do |subscription|
    next unless result[subscription.id]

    subscription.assign_attributes(updated_at: result[subscription.id])
  end
end

.subscription_queries ⇒ `PgEventstore::SubscriptionQueries`



104
105
106

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 104

def subscription_queries
  SubscriptionQueries.new(connection, QueryStrategy::Foreground.new(connection))
end

Instance Method Details

#==(other) ⇒ `Boolean`

@param another

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 165

def ==(other)
  return false unless other.is_a?(Subscription)

  id == other.id
end

#assign_attributes(attrs) ⇒ `::Hash[untyped, untyped]`

@param attrs

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 121

def assign_attributes(attrs)
  attrs.each do |attr, value|
    public_send("#{attr}=", value)
  end
end

#dup ⇒ `PgEventstore::Subscription`

Dup the current object without assigned connection



140
141
142

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 140

def dup
  Subscription.new(**Utils.deep_dup(options_hash))
end

#eql?(other) ⇒ `Boolean`

@param another

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 157

def eql?(other)
  return false unless other.is_a?(Subscription)

  hash == other.hash
end

#hash ⇒ `Integer`



151
152
153

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 151

def hash
  id.hash
end

#init_default_values ⇒ `void`

This method returns an undefined value.

@param options

58	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 58 def init_default_values: (::Hash[untyped, untyped] options) -> void

#lock!(lock_id, force: false) ⇒ `PgEventstore::Subscription`

@param lock_id

@param force

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 131

def lock!(lock_id, force: false)
  self.id = subscription_queries.find_or_create_by(set:, name:)[:id]
  self.locked_by = subscription_queries.lock!(id, lock_id, force:)
  reset_runtime_attributes
  self
end

#options_hash ⇒ `::Hash[untyped, untyped]`

46	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 46 def options_hash: () -> ::Hash[untyped, untyped]

#readonly! ⇒ `Boolean`

@param opt_name

49	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 49 def readonly!: (Symbol opt_name) -> bool

#readonly? ⇒ `Boolean`

@param opt_name

52	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 52 def readonly?: (Symbol opt_name) -> bool

#readonly_error ⇒ `void`

This method returns an undefined value.

@param opt_name

55	# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 55 def readonly_error: (Symbol opt_name) -> void

#reload ⇒ `PgEventstore::Subscription`

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 145

def reload
  assign_attributes(subscription_queries.find!(id))
  self
end

#reset_runtime_attributes ⇒ `void`

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 174

def reset_runtime_attributes
  update(
    options:,
    restart_count: 0,
    last_restarted_at: nil,
    max_restarts_number:,
    chunk_query_interval: [chunk_query_interval, MIN_EVENTS_PULL_INTERVAL].max,
    last_chunk_fed_at: DEFAULT_TIMESTAMP,
    last_chunk_greatest_position: nil,
    last_error: nil,
    last_error_occurred_at: nil,
    time_between_restarts:,
    state: RunnerState::STATES[:initial]
  )
  subscription_queries.create_or_replace_table_function(id, options, locked_by)
  reload
end

#subscription_queries ⇒ `PgEventstore::SubscriptionQueries`



193
194
195

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 193

def subscription_queries
  self.class.subscription_queries
end

#update(attrs) ⇒ `::Hash[untyped, untyped]`

@param attrs



115
116
117

# File 'lib/pg_eventstore/subscriptions/subscription.rb', line 115

def update(attrs)
  assign_attributes(subscription_queries.update(id, attrs:, locked_by:))
end

Class: PgEventstore::Subscription

Overview

Constant Summary collapse

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Extensions::OptionsExtension

Constructor Details

#initialize ⇒ Subscription

Instance Attribute Details

#average_event_processing_time ⇒ Float?

#chunk_query_interval ⇒ Integer, ...

#created_at ⇒ Time?

#current_position ⇒ Integer?

#id ⇒ Integer?

#last_chunk_fed_at ⇒ Time?

#last_chunk_greatest_position ⇒ Integer?

#last_error ⇒ Hash?

#last_error_occurred_at ⇒ Time?

#last_restarted_at ⇒ Time?

#locked_by ⇒ Integer?

#max_restarts_number ⇒ Integer?

#name ⇒ String?

#options ⇒ Hash?

#restart_count ⇒ Integer?

#set ⇒ String?

#state ⇒ String?

#time_between_restarts ⇒ Integer?

#total_processed_events ⇒ Integer?

#updated_at ⇒ Time?

Class Method Details

.ping_all(subscriptions_set_id, subscriptions) ⇒ void

.subscription_queries ⇒ PgEventstore::SubscriptionQueries

Instance Method Details

#==(other) ⇒ Boolean

#assign_attributes(attrs) ⇒ ::Hash[untyped, untyped]

#dup ⇒ PgEventstore::Subscription

#eql?(other) ⇒ Boolean

#hash ⇒ Integer

#init_default_values ⇒ void

#lock!(lock_id, force: false) ⇒ PgEventstore::Subscription

#options_hash ⇒ ::Hash[untyped, untyped]

#readonly! ⇒ Boolean

#readonly? ⇒ Boolean

#readonly_error ⇒ void

#reload ⇒ PgEventstore::Subscription

#reset_runtime_attributes ⇒ void

#subscription_queries ⇒ PgEventstore::SubscriptionQueries

#update(attrs) ⇒ ::Hash[untyped, untyped]

#initialize ⇒ `Subscription`

#average_event_processing_time ⇒ `Float`^?

#chunk_query_interval ⇒ `Integer`, ...

#created_at ⇒ `Time`^?

#current_position ⇒ `Integer`^?

#id ⇒ `Integer`^?

#last_chunk_fed_at ⇒ `Time`^?

#last_chunk_greatest_position ⇒ `Integer`^?

#last_error ⇒ `Hash`^?

#last_error_occurred_at ⇒ `Time`^?

#last_restarted_at ⇒ `Time`^?

#locked_by ⇒ `Integer`^?

#max_restarts_number ⇒ `Integer`^?

#name ⇒ `String`^?

#options ⇒ `Hash`^?

#restart_count ⇒ `Integer`^?

#set ⇒ `String`^?

#state ⇒ `String`^?

#time_between_restarts ⇒ `Integer`^?

#total_processed_events ⇒ `Integer`^?

#updated_at ⇒ `Time`^?

.ping_all(subscriptions_set_id, subscriptions) ⇒ `void`

.subscription_queries ⇒ `PgEventstore::SubscriptionQueries`

#==(other) ⇒ `Boolean`

#assign_attributes(attrs) ⇒ `::Hash[untyped, untyped]`

#dup ⇒ `PgEventstore::Subscription`

#eql?(other) ⇒ `Boolean`

#hash ⇒ `Integer`

#init_default_values ⇒ `void`

#lock!(lock_id, force: false) ⇒ `PgEventstore::Subscription`

#options_hash ⇒ `::Hash[untyped, untyped]`

#readonly! ⇒ `Boolean`

#readonly? ⇒ `Boolean`

#readonly_error ⇒ `void`

#reload ⇒ `PgEventstore::Subscription`

#reset_runtime_attributes ⇒ `void`

#subscription_queries ⇒ `PgEventstore::SubscriptionQueries`

#update(attrs) ⇒ `::Hash[untyped, untyped]`