Class: Kaal::Dispatch::DatabaseEngine

Inherits:
Registry
  • Object
show all
Defined in:
lib/kaal/dispatch/database_engine.rb,
sig/kaal/dispatch/database_engine.rbs

Overview

Sequel-backed dispatch registry stored in kaal_dispatches.

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Registry

#dispatched?

Constructor Details

#initialize(database:, namespace: nil) ⇒ DatabaseEngine

Returns a new instance of DatabaseEngine.

Parameters:

  • database: (Kaal::rbs_any)
  • namespace: (Kaal::rbs_any, nil) (defaults to: nil) 


14
15
16
17
18
 
# File 'lib/kaal/dispatch/database_engine.rb', line 14

def initialize(database:, namespace: nil)
  super()
  @database = Kaal::Persistence::Database.new(database)
  @namespace = namespace
end

Class Method Details

.normalize_row(row, namespace: nil) ⇒ nil, { key: Kaal::rbs_any, fire_time: Kaal::rbs_any, dispatched_at: Kaal::rbs_any, node_id: Kaal::rbs_any, status: Kaal::rbs_any }

Parameters:

  • row (Kaal::rbs_any)
  • namespace: (Kaal::rbs_any, nil) (defaults to: nil)

Returns:

  • (nil, { key: Kaal::rbs_any, fire_time: Kaal::rbs_any, dispatched_at: Kaal::rbs_any, node_id: Kaal::rbs_any, status: Kaal::rbs_any }) 


71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/kaal/dispatch/database_engine.rb', line 71

def self.normalize_row(row, namespace: nil)
  return nil unless row

  {
    key: strip_namespace(row[:key], namespace:),
    fire_time: row[:fire_time],
    dispatched_at: row[:dispatched_at],
    node_id: row[:node_id],
    status: row[:status]
  }
end

.strip_namespace(key, namespace:) ⇒ Kaal::rbs_any

Parameters:

  • key (Kaal::rbs_any)
  • namespace: (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


83
84
85
86
87
88
 
# File 'lib/kaal/dispatch/database_engine.rb', line 83

def self.strip_namespace(key, namespace:)
  return key if namespace.to_s.empty?

  prefix = "#{namespace}:"
  key.start_with?(prefix) ? key.delete_prefix(prefix) : key
end

Instance Method Details

#cleanup(recovery_window: 86_400) ⇒ Kaal::rbs_any

Parameters:

  • recovery_window: (::Integer) (defaults to: 86_400)

Returns:

  • (Kaal::rbs_any) 


66
67
68
69
 
# File 'lib/kaal/dispatch/database_engine.rb', line 66

def cleanup(recovery_window: 86_400)
  cutoff_time = Time.now.utc - recovery_window
  cleanup_dataset.where { fire_time < cutoff_time }.delete
end

#cleanup_dataset ⇒ Kaal::rbs_any

Returns:

  • (Kaal::rbs_any) 


113
114
115
116
117
 
# File 'lib/kaal/dispatch/database_engine.rb', line 113

def cleanup_dataset
  return dataset if @namespace.to_s.empty?

  dataset.where(::Sequel.lit('key LIKE ?', "#{@namespace}:%"))
end

#dataset ⇒ Kaal::rbs_any

Returns:

  • (Kaal::rbs_any) 


92
93
94
 
# File 'lib/kaal/dispatch/database_engine.rb', line 92

def dataset
  @database.dispatches_dataset
end

#find_by_key(key) ⇒ Kaal::rbs_any

Parameters:

  • key (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


54
55
56
 
# File 'lib/kaal/dispatch/database_engine.rb', line 54

def find_by_key(key)
  query(key: namespaced_key(key))
end

#find_by_node(node_id) ⇒ Kaal::rbs_any

Parameters:

  • node_id (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


58
59
60
 
# File 'lib/kaal/dispatch/database_engine.rb', line 58

def find_by_node(node_id)
  query(node_id: node_id)
end

#find_by_status(status) ⇒ Kaal::rbs_any

Parameters:

  • status (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


62
63
64
 
# File 'lib/kaal/dispatch/database_engine.rb', line 62

def find_by_status(status)
  query(status: status)
end

#find_dispatch(key, fire_time) ⇒ Kaal::rbs_any

Parameters:

  • key (Kaal::rbs_any)
  • fire_time (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


50
51
52
 
# File 'lib/kaal/dispatch/database_engine.rb', line 50

def find_dispatch(key, fire_time)
  self.class.normalize_row(dataset.where(key: namespaced_key(key), fire_time: fire_time).first, namespace: @namespace)
end

#log_dispatch(key, fire_time, node_id, status = 'dispatched') ⇒ Kaal::rbs_any

Parameters:

  • key (Kaal::rbs_any)
  • fire_time (Kaal::rbs_any)
  • node_id (Kaal::rbs_any)
  • status (::String) (defaults to: 'dispatched')

Returns:

  • (Kaal::rbs_any) 


20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/kaal/dispatch/database_engine.rb', line 20

def log_dispatch(key, fire_time, node_id, status = 'dispatched')
  now = Time.now.utc
  storage_key = namespaced_key(key)
  attributes = {
    key: storage_key,
    fire_time: fire_time,
    dispatched_at: now,
    node_id: node_id,
    status: status
  }
  dispatches_dataset = dataset
  update_values = { dispatched_at: now, node_id: node_id, status: status }
  begin
    dispatches_dataset.insert_conflict(
      target: i[key fire_time],
      update: update_values
    ).insert(attributes)
  rescue NoMethodError => e
    raise unless e.name == :insert_conflict

    begin
      dispatches_dataset.insert(attributes)
    rescue ::Sequel::UniqueConstraintViolation
      dispatches_dataset.where(key: storage_key, fire_time: fire_time).update(update_values)
    end
  end

  find_dispatch(key, fire_time)
end

#namespaced_key(key) ⇒ Kaal::rbs_any, ::String

Parameters:

  • key (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any, ::String) 


96
97
98
99
100
 
# File 'lib/kaal/dispatch/database_engine.rb', line 96

def namespaced_key(key)
  return key if @namespace.to_s.empty?

  "#{@namespace}:#{key}"
end

#query(filters) ⇒ Kaal::rbs_any

Parameters:

  • filters (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


102
103
104
 
# File 'lib/kaal/dispatch/database_engine.rb', line 102

def query(filters)
  query_dataset(filters).reverse_order(:fire_time).all.map { |row| self.class.normalize_row(row, namespace: @namespace) }
end

#query_dataset(filters) ⇒ Kaal::rbs_any

Parameters:

  • filters (Kaal::rbs_any)

Returns:

  • (Kaal::rbs_any) 


106
107
108
109
110
111
 
# File 'lib/kaal/dispatch/database_engine.rb', line 106

def query_dataset(filters)
  relation = dataset.where(filters)
  return relation if @namespace.to_s.empty? || filters.key?(:key)

  relation.where(::Sequel.lit('key LIKE ?', "#{@namespace}:%"))
end