Merge 111d9a1 into d2293c9

Author: bedrock-adam

README.md

@@ -51,13 +51,15 @@ end
 ```ruby
 # bin/outboxer_publisher
 
-Outboxer::Publisher.publish_message(...) do |message|
-  # TODO: publish message here
-
-  logger.info "Outboxer published message " \
-    "id=#{message[:id]} " \
-    "messageable_type=#{message[:messageable_type]} " \
-    "messageable_id=#{message[:messageable_id]} "
+Outboxer::Publisher.publish_messages do |messages|
+  # TODO: publish messages here
+
+  messages.each do |message|
+    logger.info "Outboxer published message " \
+      "id=#{message[:id]} " \
+      "messageable_type=#{message[:messageable_type]} " \
+      "messageable_id=#{message[:messageable_id]} "
+  end
 end
 ```
 

bin/outboxer_publisher

@@ -7,7 +7,7 @@ cli_options = Outboxer::Publisher.parse_cli_options(ARGV)
 environment = cli_options.delete(:environment) || ENV["APP_ENV"] || ENV["RAILS_ENV"] || "development"
 config_path = cli_options.delete(:config) || Outboxer::Publisher::CONFIG_DEFAULTS[:path]
 config = Outboxer::Publisher.config(environment: environment, path: config_path)
-options = Outboxer::Publisher::PUBLISH_MESSAGE_DEFAULTS.merge(config.merge(cli_options))
+options = Outboxer::Publisher::PUBLISH_MESSAGES_DEFAULTS.merge(config.merge(cli_options))
 logger = Outboxer::Logger.new($stdout, level: options[:log_level])
 
 database_config = Outboxer::Database.config(
@@ -16,7 +16,7 @@ database_config = Outboxer::Database.config(
 Outboxer::Database.connect(config: database_config, logger: logger)
 
 begin
-  Outboxer::Publisher.publish_message(
+  Outboxer::Publisher.publish_messages(
     batch_size: options[:batch_size],
     buffer_size: options[:buffer_size],
     concurrency: options[:concurrency],

lib/outboxer/publisher.rb

@@ -88,8 +88,8 @@ def config(
 
       yaml = YAML.safe_load(erb_result, permitted_classes: [Symbol], aliases: true)
       yaml.deep_symbolize_keys!
-      yaml_override = yaml.fetch(environment&.to_sym, {}).slice(*PUBLISH_MESSAGE_DEFAULTS.keys)
-      yaml.slice(*PUBLISH_MESSAGE_DEFAULTS.keys).merge(yaml_override)
+      yaml_override = yaml.fetch(environment&.to_sym, {}).slice(*PUBLISH_MESSAGES_DEFAULTS.keys)
+      yaml.slice(*PUBLISH_MESSAGES_DEFAULTS.keys).merge(yaml_override)
     rescue Errno::ENOENT
       {}
     end
@@ -499,7 +499,7 @@ def handle_signal(id:, name:, logger:)
       end
     end
 
-    PUBLISH_MESSAGE_DEFAULTS = {
+    PUBLISH_MESSAGES_DEFAULTS = {
       batch_size: 10,
       buffer_size: 100,
       concurrency: 1,
@@ -522,15 +522,15 @@ def handle_signal(id:, name:, logger:)
     # @param process [Process] The process module for system metrics.
     # @param kernel [Kernel] The kernel module for sleeping operations.
     # @yield [Hash] A block to handle the publishing of each message.
-    def publish_message(
+    def publish_messages(
       name: "#{::Socket.gethostname}:#{::Process.pid}",
-      batch_size: PUBLISH_MESSAGE_DEFAULTS[:batch_size],
-      buffer_size: PUBLISH_MESSAGE_DEFAULTS[:buffer_size],
-      concurrency: PUBLISH_MESSAGE_DEFAULTS[:concurrency],
-      tick_interval: PUBLISH_MESSAGE_DEFAULTS[:tick_interval],
-      poll_interval: PUBLISH_MESSAGE_DEFAULTS[:poll_interval],
-      heartbeat_interval: PUBLISH_MESSAGE_DEFAULTS[:heartbeat_interval],
-      logger: Logger.new($stdout, level: PUBLISH_MESSAGE_DEFAULTS[:log_level]),
+      batch_size: PUBLISH_MESSAGES_DEFAULTS[:batch_size],
+      buffer_size: PUBLISH_MESSAGES_DEFAULTS[:buffer_size],
+      concurrency: PUBLISH_MESSAGES_DEFAULTS[:concurrency],
+      tick_interval: PUBLISH_MESSAGES_DEFAULTS[:tick_interval],
+      poll_interval: PUBLISH_MESSAGES_DEFAULTS[:poll_interval],
+      heartbeat_interval: PUBLISH_MESSAGES_DEFAULTS[:heartbeat_interval],
+      logger: Logger.new($stdout, level: PUBLISH_MESSAGES_DEFAULTS[:log_level]),
       time: ::Time, process: ::Process, kernel: ::Kernel,
       &block
     )

spec/lib/outboxer/publisher/publish_message_spec.rb renamed to spec/lib/outboxer/publisher/publish_messages_spec.rb

@@ -3,7 +3,7 @@
 
 module Outboxer
   RSpec.describe Publisher do
-    describe ".publish_message" do
+    describe ".publish_messages" do
       let(:batch_size) { 1 }
       let(:buffer_size) { 1 }
       let(:poll_interval) { 1 }
@@ -19,13 +19,13 @@ module Outboxer
 
       context "when TTIN signal sent" do
         it "dumps stack trace" do
-          publish_message_thread = Thread.new do
-            Outboxer::Publisher.publish_message(
+          publish_messages_thread = Thread.new do
+            Outboxer::Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,
               tick_interval: tick_interval,
-              logger: logger, kernel: kernel) do |_message|
+              logger: logger, kernel: kernel) do |_messages|
               ::Process.kill("TTIN", ::Process.pid)
             end
           end
@@ -34,7 +34,7 @@ module Outboxer
 
           ::Process.kill("TERM", ::Process.pid)
 
-          publish_message_thread.join
+          publish_messages_thread.join
 
           expect(logger)
             .to have_received(:info)
@@ -45,14 +45,14 @@ module Outboxer
 
       context "when stopped and resumed during message publishing" do
         it "stops and resumes the publishing process correctly" do
-          publish_message_thread = Thread.new do
-            Outboxer::Publisher.publish_message(
+          publish_messages_thread = Thread.new do
+            Outboxer::Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,
               tick_interval: tick_interval,
               logger: logger,
-              kernel: kernel) do |_message|
+              kernel: kernel) do |_messages|
               ::Process.kill("TSTP", ::Process.pid)
             end
           end
@@ -63,13 +63,13 @@ module Outboxer
 
           ::Process.kill("TERM", ::Process.pid)
 
-          publish_message_thread.join
+          publish_messages_thread.join
         end
       end
 
       context "when message published successfully" do
         it "sets the message to published" do
-          Publisher.publish_message(
+          Publisher.publish_messages(
             batch_size: batch_size,
             buffer_size: buffer_size,
             poll_interval: poll_interval,
@@ -95,7 +95,7 @@ module Outboxer
           let(:standard_error) { StandardError.new("some error") }
 
           before do
-            Publisher.publish_message(
+            Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,
@@ -120,7 +120,7 @@ module Outboxer
             expect(queued_message.exceptions[0].frames[0].index).to eq(0)
 
             expect(queued_message.exceptions[0].frames[0].text).to match(
-              /outboxer\/publisher\/publish_message_spec.rb:\d+:in `block \(6 levels\) in <module:Outboxer>'/)
+              /outboxer\/publisher\/publish_messages_spec.rb:\d+:in `block \(6 levels\) in <module:Outboxer>'/)
           end
 
           it "logs errors" do
@@ -137,7 +137,7 @@ module Outboxer
           let(:no_memory_error) { NoMemoryError.new }
 
           before do
-            Publisher.publish_message(
+            Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,
@@ -159,7 +159,7 @@ module Outboxer
 
             expect(queued_message.exceptions[0].frames[0].index).to eq(0)
             expect(queued_message.exceptions[0].frames[0].text).to match(
-              /outboxer\/publisher\/publish_message_spec.rb:\d+:in `block \(6 levels\) in <module:Outboxer>'/)
+              /outboxer\/publisher\/publish_messages_spec.rb:\d+:in `block \(6 levels\) in <module:Outboxer>'/)
           end
 
           it "logs errors" do
@@ -191,7 +191,7 @@ module Outboxer
 
             expect(logger).to receive(:error).with(include("StandardError: queue error")).once
 
-            Publisher.publish_message(
+            Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,
@@ -210,7 +210,7 @@ module Outboxer
               .with(include("NoMemoryError: failed to allocate memory"))
               .once
 
-            Publisher.publish_message(
+            Publisher.publish_messages(
               batch_size: batch_size,
               buffer_size: buffer_size,
               poll_interval: poll_interval,