Class: RocketJob::Event

Inherits:
Object
  • Object
show all
Includes:
Mongoid::Timestamps, Plugins::Document, SemanticLogger::Loggable
Defined in:
lib/rocket_job/event.rb

Overview

RocketJob::Event

Publish and Subscribe to events. Events are published immediately and usually consumed almost immediately by all subscriber processes.

Constant Summary collapse

ALL_EVENTS = 
"*".freeze
IGNORABLE_CREATE_CODES =

MongoDB OperationFailure codes that are safe to ignore when concurrently creating the polling collection and its TTL index:

48 NamespaceExists, 85 IndexOptionsConflict, 86 IndexKeySpecsConflict.
[48, 85, 86].freeze

Class Method Summary collapse

Class Method Details

.add_subscriber(subscriber) ⇒ Object 



146
147
148
149
150
 
# File 'lib/rocket_job/event.rb', line 146

def self.add_subscriber(subscriber)
  name               = subscriber.class.event_name
  @subscribers[name] = @subscribers[name] << subscriber
  subscriber.object_id
end

.collection_exists? ⇒ Boolean

Returns:

  • (Boolean) 


232
233
234
 
# File 'lib/rocket_job/event.rb', line 232

def self.collection_exists?
  collection.database.collection_names.include?(collection_name.to_s)
end

.convert_to_capped_collection(size) ⇒ Object

Convert a non-capped collection to capped



237
238
239
 
# File 'lib/rocket_job/event.rb', line 237

def self.convert_to_capped_collection(size)
  collection.database.command("convertToCapped" => collection_name.to_s, "size" => size)
end

.create_capped_collection(size: capped_collection_size) ⇒ Object

Create the capped collection only if it does not exist. Drop the collection before calling this method to re-create it.



135
136
137
138
139
140
141
 
# File 'lib/rocket_job/event.rb', line 135

def self.create_capped_collection(size: capped_collection_size)
  if collection_exists?
    convert_to_capped_collection(size) unless collection.capped?
  else
    collection.client[collection_name, {capped: true, size: size}].create
  end
end

.create_polling_collection ⇒ Object

Create the regular (non-capped) collection used by the polling strategy, along with a TTL index that expires old events.

Safe to call repeatedly: it never drops data, and tolerates the collection and index already existing.



178
179
180
181
182
183
184
185
 
# File 'lib/rocket_job/event.rb', line 178

def self.create_polling_collection
  collection.client[collection_name].create unless collection_exists?
  collection.indexes.create_one({created_at: 1}, expire_after: event_retention_seconds)
rescue Mongo::Error::OperationFailure => e
  # Ignore "collection already exists" and "index already exists" races between
  # multiple servers starting at once. Anything else is a real error.
  raise(e) unless IGNORABLE_CREATE_CODES.include?(e.code)
end

.listener(time: @load_time) ⇒ Object

Indefinitely watch for new events, dispatching each to its subscribers. time: the start time from which to start looking for new events.



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/rocket_job/event.rb', line 113

def self.listener(time: @load_time)
  Thread.current.name = "rocketjob event"

  case listener_strategy
  when :capped
    create_capped_collection
    logger.info("Event listener started (capped collection)")
    tail_capped_collection(time) { |event| process_event(event) }
  when :polling
    create_polling_collection
    logger.info("Event listener started (polling, interval: #{poll_interval}s)")
    poll_collection(time) { |event| process_event(event) }
  else
    raise(ArgumentError, "Unknown RocketJob::Event.listener_strategy: #{listener_strategy.inspect}")
  end
rescue Exception => e
  logger.error("#listener Event listener is terminating due to unhandled exception", e)
  raise(e)
end

.poll_collection(time, &block) ⇒ Object

Indefinitely poll a regular collection for new events. time: the start time from which to start looking for new events.

After the first poll the _id of the last event seen is used as the watermark, which is monotonic and unique, so events sharing a created_at timestamp are never skipped.



193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/rocket_job/event.rb', line 193

def self.poll_collection(time, &block)
  last_id = nil
  loop do
    last_id = poll_once(time, last_id, &block)
    sleep(poll_interval)
  end
rescue Mongo::Error::SocketError, Mongo::Error::SocketTimeoutError, Mongo::Error::OperationFailure, Timeout::Error => e
  logger.info("Polling failed, retrying: #{e.class.name} #{e.message}")
  sleep(poll_interval)
  retry
end

.poll_once(time, last_id = nil) ⇒ Object

Perform a single polling pass, yielding every event newer than the watermark and returning the new watermark (the _id of the last event seen, or the supplied last_id when no new events were found).



208
209
210
211
212
213
214
215
 
# File 'lib/rocket_job/event.rb', line 208

def self.poll_once(time, last_id = nil)
  filter = last_id ? {_id: {"$gt" => last_id}} : {created_at: {"$gt" => time}}
  collection.find(filter).sort(_id: 1).each do |doc|
    last_id = doc["_id"]
    yield(Mongoid::Factory.from_db(Event, doc))
  end
  last_id
end

.process_event(event) ⇒ Object

Process a new event, calling registered subscribers.



218
219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/rocket_job/event.rb', line 218

def self.process_event(event)
  logger.info("Event Received", event.attributes)

  if @subscribers.key?(event.name)
    @subscribers[event.name].each { |subscriber| subscriber.process_action(event.action, event.parameters) }
  end

  if @subscribers.key?(ALL_EVENTS)
    @subscribers[ALL_EVENTS].each { |subscriber| subscriber.process_event(event.name, event.action, event.parameters) }
  end
rescue StandardError => e
  logger.error("Unknown subscriber. Continuing..", e)
end

.subscribe(subscriber) ⇒ Object

Add a subscriber for its events. Returns a handle to the subscription that can be used to unsubscribe this particular subscription

Example: def MySubscriber include RocketJob::Subscriber

def hello
logger.info "Hello Action Received"
end

def show(message:)
logger.info "Received: #{message}"
end

end

MySubscriber.subscribe



93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/rocket_job/event.rb', line 93

def self.subscribe(subscriber)
  if block_given?
    begin
      handle = add_subscriber(subscriber)
      yield(subscriber)
    ensure
      unsubscribe(handle) if handle
    end
  else
    add_subscriber(subscriber)
  end
end

.tail_capped_collection(time) ⇒ Object 



152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/rocket_job/event.rb', line 152

def self.tail_capped_collection(time)
  with(socket_timeout: long_poll_seconds + 10) do
    filter = {created_at: {"$gt" => time}}
    collection.
      find(filter).
      await_data.
      cursor_type(:tailable_await).
      max_await_time_ms(long_poll_seconds * 1000).
      sort("$natural" => 1).
      each do |doc|
      event = Mongoid::Factory.from_db(Event, doc)
      # Recovery will occur from after the last message read
      time = event.created_at
      yield(event)
    end
  end
rescue Mongo::Error::SocketError, Mongo::Error::SocketTimeoutError, Mongo::Error::OperationFailure, Timeout::Error => e
  logger.info("Creating a new cursor and trying again: #{e.class.name} #{e.message}")
  retry
end

.unsubscribe(handle) ⇒ Object

Unsubscribes a previous subscription



107
108
109
 
# File 'lib/rocket_job/event.rb', line 107

def self.unsubscribe(handle)
  @subscribers.each_value { |v| v.delete_if { |i| i.object_id == handle } }
end