Class: Kaal::Internal::ActiveRecord::MySQLBackend
Overview
MySQL named-lock engine paired with Active Record registries.
Constant Summary
collapse
- MAX_LOCK_NAME_LENGTH =
64
- UNSET_SKIP_LOCKED_SUPPORT =
Object.new.freeze
Class Method Summary
collapse
Instance Method Summary
collapse
#dispatch_attempt_logger, #log_dispatch_attempt, #parse_lock_key, parse_lock_key
#disconnect_for_fork, #with_lock
Constructor Details
#initialize(connection = nil, dispatch_registry: nil, definition_registry: nil, namespace: nil, use_skip_locked: UNSET_SKIP_LOCKED_SUPPORT) ⇒ MySQLBackend
Returns a new instance of MySQLBackend.
20
21
22
23
24
25
26
27
28
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 20
def initialize(connection = nil, dispatch_registry: nil, definition_registry: nil, namespace: nil,
use_skip_locked: UNSET_SKIP_LOCKED_SUPPORT)
super()
ConnectionSupport.configure!(connection)
@dispatch_registry = dispatch_registry
@definition_registry = definition_registry
@namespace = namespace
@use_skip_locked = use_skip_locked
end
|
Class Method Details
.normalize_lock_name(key) ⇒ Kaal::rbs_any, ::String
56
57
58
59
60
61
62
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 56
def self.normalize_lock_name(key)
return key if key.length <= MAX_LOCK_NAME_LENGTH
digest = Digest::SHA256.hexdigest(key)
prefix_length = MAX_LOCK_NAME_LENGTH - 17
"#{key[0...prefix_length]}:#{digest[0...16]}"
end
|
Instance Method Details
#acquire(key, _ttl) ⇒ Kaal::rbs_any
42
43
44
45
46
47
48
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 42
def acquire(key, _ttl)
acquired = scalar('SELECT GET_LOCK(?, 0) AS lock_result', self.class.normalize_lock_name(key)) == 1
log_dispatch_attempt(key) if acquired
acquired
rescue StandardError => e
raise Kaal::Backend::LockAdapterError, "MySQL acquire failed for #{key}: #{e.message}"
end
|
#definition_registry ⇒ Kaal::rbs_any
34
35
36
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 34
def definition_registry
@definition_registry ||= DefinitionRegistry.new
end
|
#delayed_store ⇒ Kaal::rbs_any
38
39
40
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 38
def delayed_store
@delayed_store ||= DelayedJobRegistry.new(use_skip_locked: supports_skip_locked?)
end
|
#dispatch_registry ⇒ Kaal::rbs_any
30
31
32
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 30
def dispatch_registry
@dispatch_registry ||= DispatchRegistry.new(namespace: resolved_namespace)
end
|
#release(key) ⇒ Kaal::rbs_any
50
51
52
53
54
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 50
def release(key)
scalar('SELECT RELEASE_LOCK(?) AS lock_result', self.class.normalize_lock_name(key)) == 1
rescue StandardError => e
raise Kaal::Backend::LockAdapterError, "MySQL release failed for #{key}: #{e.message}"
end
|
#resolved_namespace ⇒ Kaal::rbs_any
76
77
78
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 76
def resolved_namespace
@namespace || Kaal.configuration.namespace
end
|
#scalar(sql, *binds) ⇒ Kaal::rbs_any
66
67
68
69
70
71
72
73
74
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 66
def scalar(sql, *binds)
sanitized_sql = if binds.empty?
sql
else
BaseRecord.send(:sanitize_sql_array, [sql, *binds])
end
result = BaseRecord.connection.exec_query(sanitized_sql)
result.first.values.first
end
|
#supports_skip_locked? ⇒ Boolean
80
81
82
83
84
85
|
# File 'lib/kaal/internal/active_record/mysql_backend.rb', line 80
def supports_skip_locked?
return @use_skip_locked unless @use_skip_locked.equal?(UNSET_SKIP_LOCKED_SUPPORT)
version_string = scalar('SELECT VERSION() AS version')
Kaal::DelayedJob::MySQLVersionSupport.skip_locked_supported?(version_string)
end
|