Class: Dionysus::Producer

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

Defined Under Namespace

Classes: BaseResponder, Config, DeletedRecordSerializer, Genesis, KarafkaResponderGenerator, Key, ModelSerializer, Outbox, PartitionKey, Registry, Serializer, Suppressor

Constant Summary collapse

MAX_SNAPSHOT_ATTEMPTS =

a contended record can move again while it is being re-serialized; bound the work and publish the last attempt rather than looping or dropping the message. Overridable per producer through config.max_snapshot_attempts - this is only the default.

3

Class Method Summary collapse

Class Method Details

.configuration ⇒ Object 



9
10
11
 
# File 'lib/dionysus/producer.rb', line 9

def self.configuration
  @configuration ||= Dionysus::Producer::Config.new
end

.configure {|configuration| ... } ⇒ Object

Yields:



13
14
15
 
# File 'lib/dionysus/producer.rb', line 13

def self.configure
  yield configuration
end

.declare ⇒ Object 



21
22
23
24
25
26
27
28
 
# File 'lib/dionysus/producer.rb', line 21

def self.declare(&)
  registry = Dionysus::Producer::Registry.new

  registry.instance_eval(&)
  configure do |configuration|
    configuration.registry = registry
  end
end

.observers_with_responders_for(resource, changeset) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/dionysus/producer.rb', line 106

def self.observers_with_responders_for(resource, changeset)
  return [] if registry.nil?

  registry.registrations.values.each.with_object([]) do |registration, accum|
    registration.topics.each do |topic|
      topic
        .models
        .select { |model_registration| model_registration.observes?(resource, changeset) }
        .each do |model_registration|
          association_name = model_registration.association_name_for_observable(resource, changeset)
          methods_chain = association_name.to_s.split(".")
          association_or_associations = methods_chain.inject(resource) do |record, method_name|
            record.public_send(method_name)
          end

          accum << [Array.wrap(association_or_associations).compact, topic.producer]
        end
    end
  end
end

.outbox ⇒ Object 



30
31
32
 
# File 'lib/dionysus/producer.rb', line 30

def self.outbox
  Dionysus::Producer::Outbox.new(configuration.outbox_model, config: configuration)
end

.outbox_publisher ⇒ Object 



34
35
36
 
# File 'lib/dionysus/producer.rb', line 34

def self.outbox_publisher
  Dionysus::Producer::Outbox::Publisher.new(config: configuration)
end

.registry ⇒ Object 



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

def self.registry
  configuration.registry
end

.reset! ⇒ Object 



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

def self.reset!
  return if registry.nil?

  registry.registrations.values.flat_map(&:producers).each do |producer_class|
    Dionysus.send(:remove_const, producer_class.name.demodulize.to_sym) if producer_class.name
  end
  @configuration = Dionysus::Producer::Config.new
end

.responders_for(model_klass) ⇒ Object 



47
48
49
50
51
52
53
54
55
 
# File 'lib/dionysus/producer.rb', line 47

def self.responders_for(model_klass)
  return [] if registry.nil?

  registry.registrations.each.with_object([]) do |(_, registration), responders|
    registration.producers.select { |producer| producer.publisher_of?(model_klass) }.each do |producer|
      responders << producer
    end
  end
end

.responders_for_dependency_parent(model_klass) ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/dionysus/producer.rb', line 61

def self.responders_for_dependency_parent(model_klass)
  return [] if registry.nil?

  registry.registrations.values.each.with_object([]) do |registration, accum|
    registration.topics.each do |topic|
      topic
        .models
        .select { |model_registration| model_registration.options[:with].to_a.include?(model_klass) }
        .each do |model_registration|
        accum << [model_registration.model_klass, topic.producer]
      end
    end
  end
end

.responders_for_dependency_parent_for_topic(model_klass, topic) ⇒ Object 



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

def self.responders_for_dependency_parent_for_topic(model_klass, topic)
  responders_for_dependency_parent(model_klass).select do |_model, responder|
    responder.publisher_for_topic?(topic)
  end
end

.responders_for_model_for_topic(model_klass, topic) ⇒ Object 



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

def self.responders_for_model_for_topic(model_klass, topic)
  responders_for(model_klass).select { |responder| responder.publisher_of_model_for_topic?(model_klass, topic) }
end

.start_outbox_worker(threads_number:) ⇒ Object 



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

def self.start_outbox_worker(threads_number:)
  runners = (1..threads_number).map do
    Dionysus::Producer::Outbox::Runner.new(config: configuration)
  end
  executor = Sigurd::Executor.new(runners, sleep_seconds: 5, logger: Dionysus.logger)
  signal_handler = Sigurd::SignalHandler.new(executor)
  signal_handler.run!
end

.topics_models_mapping ⇒ Object 



91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/dionysus/producer.rb', line 91

def self.topics_models_mapping
  return {} if registry.nil?

  registry
    .registrations
    .values
    .flat_map(&:topics)
    .to_h do |topic|
      [
        topic.to_s,
        topic.models.to_h { |registration| [registration.model_klass, registration.options.fetch(:with, [])] }
      ]
    end
end