Class: Upkeep::Subscriptions::Store

Inherits:
BaseStore
  • Object
show all
Defined in:
lib/upkeep/subscriptions/store.rb

Constant Summary collapse

PERSIST_NOTIFICATION = 
"persist_subscription_store.upkeep"

Constants inherited from BaseStore

BaseStore::DEFAULT_SUBSCRIPTION_TTL, BaseStore::PRUNE_NOTIFICATION, BaseStore::TRIM_BATCH_LIMIT, BaseStore::TRIM_EVERY

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from BaseStore

#explain, #fetch, #touch, #unregister

Constructor Details

#initialize(reverse_index: ReverseIndex.new) ⇒ Store

Returns a new instance of Store.



184
185
186
187
188
189
190
 
# File 'lib/upkeep/subscriptions/store.rb', line 184

def initialize(reverse_index: ReverseIndex.new)
  @active_registry = ActiveRegistry.new(reverse_index: reverse_index)
  @pending_registry = ActiveRegistry.new
  @pending_index_entries = {}
  @reverse_index = MemoryReverseIndex.new(active_registry: active_registry, pending_registry: pending_registry)
  @next_id = 0
end

Instance Attribute Details

#reverse_index ⇒ Object (readonly)

Returns the value of attribute reverse_index.



182
183
184
 
# File 'lib/upkeep/subscriptions/store.rb', line 182

def reverse_index
  @reverse_index
end

Instance Method Details

#activate(id) ⇒ Object 



211
212
213
214
215
 
# File 'lib/upkeep/subscriptions/store.rb', line 211

def activate(id)
  with_optional_notification(PERSIST_NOTIFICATION, memory_persist_payload(operation: :persist_index)) do |payload|
    activate_subscription(id, payload: payload)
  end
end

#prune_stale!(older_than: stale_threshold, limit: nil) ⇒ Object 



200
201
202
203
204
205
206
207
208
209
 
# File 'lib/upkeep/subscriptions/store.rb', line 200

def prune_stale!(older_than: stale_threshold, limit: nil)
  stale_ids = subscriptions.filter_map do |subscription|
    id = subscription.id
    id if last_seen_at(subscription) && last_seen_at(subscription) < older_than
  end
  stale_ids = stale_ids.first(limit) if limit

  unregister(stale_ids)
  stale_ids.size
end

#register(subscriber_id:, recorder:, metadata: {}, entries: nil) ⇒ Object 



192
193
194
195
196
197
198
 
# File 'lib/upkeep/subscriptions/store.rb', line 192

def register(subscriber_id:, recorder:, metadata: {}, entries: nil)
  subscription = with_optional_notification(PERSIST_NOTIFICATION, memory_persist_payload(operation: :persist_subscription)) do |payload|
    register_subscription(subscriber_id: subscriber_id, recorder: recorder, metadata: metadata, entries: entries, payload: payload)
  end
  trim_opportunistically
  subscription
end

#reset ⇒ Object 



225
226
227
228
229
230
231
 
# File 'lib/upkeep/subscriptions/store.rb', line 225

def reset
  @active_registry = ActiveRegistry.new
  @pending_registry = ActiveRegistry.new
  @pending_index_entries = {}
  @reverse_index = MemoryReverseIndex.new(active_registry: active_registry, pending_registry: pending_registry)
  @next_id = 0
end

#shutdown ⇒ Object 



217
218
219
 
# File 'lib/upkeep/subscriptions/store.rb', line 217

def shutdown
  true
end

#subscriptions ⇒ Object 



221
222
223
 
# File 'lib/upkeep/subscriptions/store.rb', line 221

def subscriptions
  active_registry.subscriptions + pending_registry.subscriptions
end

#summary ⇒ Object 



233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
 
# File 'lib/upkeep/subscriptions/store.rb', line 233

def summary
  active = active_registry.summary
  pending = pending_registry.summary
  {
    subscriptions: subscriptions.size,
    pending_subscriptions: pending_registry.count,
    active_subscriptions: active_registry.count,
    deferred_index_subscriptions: 0,
    reverse_index: active.merge(
      mode: :active,
      active: active,
      pending: pending,
      persistent: { lookup_keys: 0, entries: 0 }
    )
  }
end