Class: PgEventstore::Subscription

Inherits:
Object
  • Object
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.

Returns:

  • (Float)

See Also:

0.2
DEFAULT_TIMESTAMP =

Returns:

  • (Time) 
Time.at(0).utc.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Extensions::OptionsExtension

included

Constructor Details

#initialize ⇒ Subscription

Returns a new instance of Subscription.

Parameters:

  • options (Object) 


44
 
# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 44

def initialize: (**untyped options) -> void

Instance Attribute Details

#average_event_processing_time ⇒ Float?

Returns a speed of the subscription. Divide 1 by this value to determine how much events are processed by the Subscription per second.

Parameters:

  • value (Float, nil)

Returns:

  • (Float, nil) —

    a speed of the subscription. Divide 1 by this value to determine how much events are processed by the Subscription per second.



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

attribute(:average_event_processing_time)

#chunk_query_interval ⇒ Integer, ...

Returns determines how often to pull events for the given Subscription in seconds.

Parameters:

  • value (Integer?, Float, nil)

Returns:

  • (Integer, Float, nil) —

    determines how often to pull events for the given Subscription in seconds



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

attribute(:chunk_query_interval)

#created_at ⇒ Time?

Parameters:

  • value (Time, nil)

Returns:

  • (Time, nil) 


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

attribute(:created_at)

#current_position ⇒ Integer?

Returns current Subscription's position. It is updated automatically each time an event is processed.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    current Subscription's position. It is updated automatically each time an event is processed



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

attribute(:current_position)

#id ⇒ Integer?

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) 


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

attribute(:id)

#last_chunk_fed_at ⇒ Time?

Returns shows the time when last time events were fed to the event's processor.

Parameters:

  • value (Time, nil)

Returns:

  • (Time, nil) —

    shows the time when last time events were fed to the event's processor



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

attribute(:last_chunk_fed_at)

#last_chunk_greatest_position ⇒ Integer?

Returns shows the greatest global_position of the last event in the last chunk fed to the event's processor.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    shows the greatest global_position of the last event in the last chunk fed to the event's processor



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

attribute(:last_chunk_greatest_position)

#last_error ⇒ Hash?

Returns the information about last error caused when processing events by the Subscription.

Parameters:

  • value ({ 'class' => String, 'message' => String, 'backtrace' => Array[String] }, nil)

Returns:

  • (Hash, nil) —

    the information about last error caused when processing events by the Subscription.



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

attribute(:last_error)

#last_error_occurred_at ⇒ Time?

Returns the time when the last error occurred.

Parameters:

  • value (Time, nil)

Returns:

  • (Time, nil) —

    the time when the last error occurred



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

attribute(:last_error_occurred_at)

#last_restarted_at ⇒ Time?

Returns last time the Subscription was restarted.

Parameters:

  • value (Time, nil)

Returns:

  • (Time, nil) —

    last time the Subscription was restarted



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

attribute(:last_restarted_at)

#locked_by ⇒ Integer?

Returns The id of subscription manager which obtained the lock of the Subscription. nil value means that the Subscription isn't locked yet by any subscription manager.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    The id of subscription manager which obtained the lock of the Subscription. nil value means that the Subscription isn't locked yet by any subscription manager.



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

attribute(:locked_by)

#max_restarts_number ⇒ Integer?

Returns maximum number of times the Subscription can be restarted.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    maximum number of times the Subscription can be restarted



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

attribute(:max_restarts_number)

#name ⇒ String?

Returns Subscription's name. Subscription should have unique pair of set and name.

Parameters:

  • value (String, nil)

Returns:

  • (String, nil) —

    Subscription's name. Subscription should have unique pair of set and name.



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

attribute(:name)

#options ⇒ Hash?

Returns subscription's options to be used to query events. See SubscriptionManager#subscribe for the list of available options.

Parameters:

  • value (::Hash[untyped, untyped], nil)

Returns:

  • (Hash, nil) —

    subscription's options to be used to query events. See SubscriptionManager#subscribe for the list of available options



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

attribute(:options)

#restart_count ⇒ Integer?

Returns the number of Subscription's restarts after its failure.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    the number of Subscription's restarts after its failure



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

attribute(:restart_count)

#set ⇒ String?

Returns Subscription's set. Subscription should have unique pair of set and name.

Parameters:

  • value (String, nil)

Returns:

  • (String, nil) —

    Subscription's set. Subscription should have unique pair of set and name.



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

attribute(:set)

#state ⇒ String?

Returns current Subscription's state. It is updated automatically during Subscription's life cycle See RunnerState::STATES for possible values.

Parameters:

  • value (String, nil)

Returns:

  • (String, nil) —

    current Subscription's state. It is updated automatically during Subscription's life cycle See RunnerState::STATES for possible values.



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

attribute(:state)

#time_between_restarts ⇒ Integer?

Returns interval in seconds between retries of failed Subscription.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    interval in seconds between retries of failed Subscription



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

attribute(:time_between_restarts)

#total_processed_events ⇒ Integer?

Returns total number of events, processed by this subscription.

Parameters:

  • value (Integer, nil)

Returns:

  • (Integer, nil) —

    total number of events, processed by this subscription



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

attribute(:total_processed_events)

#updated_at ⇒ Time?

Parameters:

  • value (Time, nil)

Returns:

  • (Time, nil) 


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

Parameters:



94
95
96
97
98
99
100
101
 
# 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

Returns:

  • (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

Parameters:

  • another (Object)

Returns:

  • (Boolean) 


165
166
167
168
169
 
# 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

Parameters:

  • attrs (::Hash[untyped, untyped])

Returns:

  • (::Hash[untyped, untyped]) 


121
122
123
124
125
 
# 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

Returns:



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

Parameters:

  • another (Object)

Returns:

  • (Boolean) 


157
158
159
160
161
 
# 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

Returns:

  • (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

Parameters:

  • options (::Hash[untyped, untyped]) 


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

Parameters:

  • lock_id (Integer)
  • force: (Boolean) (defaults to: false)

Returns:



131
132
133
134
135
136
 
# 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]

Returns:

  • (::Hash[untyped, untyped]) 


46
 
# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 46

def options_hash: () -> ::Hash[untyped, untyped]

#readonly! ⇒ Boolean

@param opt_name

Parameters:

  • opt_name (Symbol)

Returns:

  • (Boolean) 


49
 
# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 49

def readonly!: (Symbol opt_name) -> bool

#readonly? ⇒ Boolean

@param opt_name

Parameters:

  • opt_name (Symbol)

Returns:

  • (Boolean) 


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

Parameters:

  • opt_name (Symbol) 


55
 
# File 'sig/pg_eventstore/subscriptions/subscription.rbs', line 55

def readonly_error: (Symbol opt_name) -> void

#reload ⇒ PgEventstore::Subscription

Returns:



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

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

#reset_runtime_attributes ⇒ void

This method returns an undefined value.



174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# 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

Returns:

  • (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

Parameters:

  • attrs (::Hash[untyped, untyped])

Returns:

  • (::Hash[untyped, untyped]) 


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