Class: Dionysus::Producer::Config

Inherits:
Object
  • Object
show all
Defined in:
lib/dionysus/producer/config.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Attribute Details

#database_connection_provider ⇒ Object

Returns the value of attribute database_connection_provider.



4
5
6
 
# File 'lib/dionysus/producer/config.rb', line 4

def database_connection_provider
  @database_connection_provider
end

#datadog_statsd_client ⇒ Object

Returns the value of attribute datadog_statsd_client.



4
5
6
 
# File 'lib/dionysus/producer/config.rb', line 4

def datadog_statsd_client
  @datadog_statsd_client
end

#default_partition_key ⇒ Object 



36
37
38
 
# File 'lib/dionysus/producer/config.rb', line 36

def default_partition_key
  @default_partition_key || :account_id
end

#error_handler ⇒ Object 



54
55
56
 
# File 'lib/dionysus/producer/config.rb', line 54

def error_handler
  @error_handler || Dionysus::Utils::NullErrorHandler
end

#event_bus ⇒ Object 



28
29
30
 
# File 'lib/dionysus/producer/config.rb', line 28

def event_bus
  @event_bus || Dionysus::Utils::NullEventBus
end

#genesis_consistency_safety_delay ⇒ Object 



76
77
78
 
# File 'lib/dionysus/producer/config.rb', line 76

def genesis_consistency_safety_delay
  @genesis_consistency_safety_delay || 60.seconds
end

#hermes_event_producer ⇒ Object 



80
81
82
 
# File 'lib/dionysus/producer/config.rb', line 80

def hermes_event_producer
  @hermes_event_producer || Dionysus::Utils::NullHermesEventProducer
end

#high_priority_sidekiq_queue ⇒ Object 



94
95
96
 
# File 'lib/dionysus/producer/config.rb', line 94

def high_priority_sidekiq_queue
  @high_priority_sidekiq_queue || self.class.high_priority_sidekiq_queue
end

#include_serialized_at_in_payload ⇒ Object

Off by default so consumers, which fall back to the offset when the field is absent, can be rolled out first.



138
139
140
141
142
 
# File 'lib/dionysus/producer/config.rb', line 138

def include_serialized_at_in_payload
  return @include_serialized_at_in_payload if defined?(@include_serialized_at_in_payload)

  false
end

#instrumenter ⇒ Object 



24
25
26
 
# File 'lib/dionysus/producer/config.rb', line 24

def instrumenter
  @instrumenter || Dionysus::Utils::NullInstrumenter
end

#lock_client ⇒ Object 



46
47
48
 
# File 'lib/dionysus/producer/config.rb', line 46

def lock_client
  @lock_client || Dionysus::Utils::NullLockClient
end

#lock_expiry_time ⇒ Object 



50
51
52
 
# File 'lib/dionysus/producer/config.rb', line 50

def lock_expiry_time
  @lock_expiry_time || 10_000
end

#max_snapshot_attempts ⇒ Object

Off by default so it can be switched on per producer, and switched back off in one env change if the extra reads ever cost more than they are worth. How many times a payload may be re-serialized before the last attempt is published as it stands. This is the expensive half of publish_consistent_snapshots: a record written faster than it serializes fails the check on every attempt, so it pays the full serialization cost this many times over and still publishes a payload that may be torn. Lower it to bound that cost; 1 disables re-serialization while leaving the check (and its metric) in place.



117
118
119
 
# File 'lib/dionysus/producer/config.rb', line 117

def max_snapshot_attempts
  @max_snapshot_attempts || Dionysus::Producer::MAX_SNAPSHOT_ATTEMPTS
end

#observers_inline_maximum_size ⇒ Object 



98
99
100
 
# File 'lib/dionysus/producer/config.rb', line 98

def observers_inline_maximum_size
  @observers_inline_maximum_size || 1000
end

#outbox_model ⇒ Object

Returns the value of attribute outbox_model.



4
5
6
 
# File 'lib/dionysus/producer/config.rb', line 4

def outbox_model
  @outbox_model
end

#outbox_publishing_batch_size ⇒ Object 



58
59
60
 
# File 'lib/dionysus/producer/config.rb', line 58

def outbox_publishing_batch_size
  @outbox_publishing_batch_size || 100
end

#outbox_worker_publishing_delay ⇒ Object 



90
91
92
 
# File 'lib/dionysus/producer/config.rb', line 90

def outbox_worker_publishing_delay
  (@outbox_worker_publishing_delay || 0).to_d.seconds
end

#outbox_worker_sleep_seconds ⇒ Object 



40
41
42
43
44
 
# File 'lib/dionysus/producer/config.rb', line 40

def outbox_worker_sleep_seconds
  return BigDecimal("0.2") if @outbox_worker_sleep_seconds.nil?

  @outbox_worker_sleep_seconds.to_d
end

#publish_after_commit ⇒ Object 



84
85
86
87
88
 
# File 'lib/dionysus/producer/config.rb', line 84

def publish_after_commit
  return @publish_after_commit if defined?(@publish_after_commit)

  false
end

#publish_consistent_snapshots ⇒ Object

Re-serialize when a record moved while its payload was being built, so the payload describes a single moment. Off by default; the extra read costs one indexed column per message, and the re-serialization only happens on the records that were actually contended.



130
131
132
133
134
 
# File 'lib/dionysus/producer/config.rb', line 130

def publish_consistent_snapshots
  return @publish_consistent_snapshots if defined?(@publish_consistent_snapshots)

  false
end

#publish_with_uncached_reads ⇒ Object 



121
122
123
124
125
 
# File 'lib/dionysus/producer/config.rb', line 121

def publish_with_uncached_reads
  return @publish_with_uncached_reads if defined?(@publish_with_uncached_reads)

  false
end

#publisher_service_name ⇒ Object 



72
73
74
 
# File 'lib/dionysus/producer/config.rb', line 72

def publisher_service_name
  @publisher_service_name || Karafka.producer.id
end

#registry ⇒ Object

Returns the value of attribute registry.



4
5
6
 
# File 'lib/dionysus/producer/config.rb', line 4

def registry
  @registry
end

#remove_consecutive_duplicates_before_publishing ⇒ Object 



144
145
146
147
148
 
# File 'lib/dionysus/producer/config.rb', line 144

def remove_consecutive_duplicates_before_publishing
  return @remove_consecutive_duplicates_before_publishing if defined?(@remove_consecutive_duplicates_before_publishing)

  false
end

#republish_deduplicated_records ⇒ Object

Publish the survivor of a collapsed run of duplicates once more, after a delay, so a payload that may have been serialized across a write is followed by one read after the writes settled. Off by default; does nothing unless remove_consecutive_duplicates_before_publishing is also on.



153
154
155
156
157
 
# File 'lib/dionysus/producer/config.rb', line 153

def republish_deduplicated_records
  return @republish_deduplicated_records if defined?(@republish_deduplicated_records)

  false
end

#republish_deduplicated_records_delay ⇒ Object

Has to outlast the burst of writes that produced the duplicates, or the republish is serialized inside the same contended window it exists to escape.



161
162
163
 
# File 'lib/dionysus/producer/config.rb', line 161

def republish_deduplicated_records_delay
  (@republish_deduplicated_records_delay || 30).to_d.seconds
end

#sidekiq_queue ⇒ Object 



68
69
70
 
# File 'lib/dionysus/producer/config.rb', line 68

def sidekiq_queue
  @sidekiq_queue || self.class.default_sidekiq_queue
end

#soft_delete_column ⇒ Object 



32
33
34
 
# File 'lib/dionysus/producer/config.rb', line 32

def soft_delete_column
  @soft_delete_column || "canceled_at"
end

#touch_records_behind_their_embedded_records ⇒ Object

Corrects a row whose own timestamp predates a record embedded in its payload, then serializes again. Off by default: it writes to the source table from the publish path.



104
105
106
107
108
 
# File 'lib/dionysus/producer/config.rb', line 104

def touch_records_behind_their_embedded_records
  return @touch_records_behind_their_embedded_records if defined?(@touch_records_behind_their_embedded_records)

  false
end

#transaction_provider ⇒ Object

Returns the value of attribute transaction_provider.



4
5
6
 
# File 'lib/dionysus/producer/config.rb', line 4

def transaction_provider
  @transaction_provider
end

#transactional_outbox_enabled ⇒ Object 



62
63
64
65
66
 
# File 'lib/dionysus/producer/config.rb', line 62

def transactional_outbox_enabled
  return @transactional_outbox_enabled if defined?(@transactional_outbox_enabled)

  true
end

Class Method Details

.default_sidekiq_queue ⇒ Object 



16
17
18
 
# File 'lib/dionysus/producer/config.rb', line 16

def self.default_sidekiq_queue
  :dionysus
end

.high_priority_sidekiq_queue ⇒ Object 



20
21
22
 
# File 'lib/dionysus/producer/config.rb', line 20

def self.high_priority_sidekiq_queue
  :dionysus_high_priority
end