Merge e919d51 into 1513c71

Author: bedrock-adam

README.md

@@ -64,7 +64,7 @@ end
 Accountify::InvoiceRaisedEvent.create!(created_at: Time.current)
 ```
 
-**8. Publish messages**
+**8. Publish outboxer messages**
 
 ```ruby
 # bin/outboxer_publisher

db/migrate/create_outboxer_messages.rb

@@ -26,12 +26,12 @@ def up
       name: "idx_outboxer_status_updated_at"
 
     # publisher latency
-    add_index :outboxer_messages, [:publisher_id, :updated_at],
-      name: "idx_outboxer_pub_id_updated_at"
+    add_index :outboxer_messages, [:publisher_id, :published_at],
+      name: "idx_outboxer_pub_id_published_at"
 
     # publisher throughput
-    add_index :outboxer_messages, [:status, :publisher_id, :updated_at],
-      name: "idx_outboxer_status_pub_id_updated_at"
+    add_index :outboxer_messages, [:status, :publisher_id, :published_at],
+      name: "idx_outboxer_status_pub_id_published_at"
 
     # bulk status + id locking
     add_index :outboxer_messages, [:status, :id],

lib/outboxer/publisher.rb

@@ -404,18 +404,19 @@ def create_heartbeat_thread(id:,
                 throughput = Models::Message
                   .where(status: Message::Status::PUBLISHED)
                   .where(publisher_id: id)
-                  .where("updated_at >= ?", 1.second.ago)
+                  .where("published_at >= ?", 1.second.ago)
                   .count
 
-                last_updated_message = Models::Message
+                last_published_message = Models::Message
+                  .where(status: Message::Status::PUBLISHED)
                   .where(publisher_id: id)
-                  .order(updated_at: :desc)
+                  .order(published_at: :desc)
                   .first
 
-                latency = if last_updated_message.nil?
+                latency = if last_published_message.nil?
                             0
                           else
-                            (Time.now.utc - last_updated_message.updated_at).to_i
+                            (Time.now.utc - last_published_message.published_at).to_i
                           end
 
                 publisher.update!(

quickstart_e2e_tests.sh

@@ -53,44 +53,45 @@ bundle exec ruby -pi -e \
   "sub(/class Event < ApplicationRecord/, \"class Event < ApplicationRecord\\n  after_create { Outboxer::Message.queue(messageable: self) }\")" \
   app/models/event.rb
 
-bundle exec rails runner 'Event.create!'
+bundle exec ruby - <<'RUBY'
+require_relative "config/environment"
 
-bundle exec ruby - <<RUBY
-publisher_cmd = File.join(Dir.pwd, "bin", "outboxer_publisher")
-read_io, write_io = IO.pipe
+event = Event.create!
 
-pid = spawn("ruby", publisher_cmd, out: write_io, err: write_io)
-write_io.close
+env = { "RAILS_ENV" => ENV["RAILS_ENV"] }
+publisher_cmd = File.join(Dir.pwd, "bin", "outboxer_publisher")
+publisher_pid = spawn(env, "ruby", publisher_cmd)
 
-output = +""
 attempt = 1
-max_attempts = 20
+max_attempts = 10
 delay = 1
 
-while attempt <= max_attempts
-  begin
-    partial = read_io.read_nonblock(1024)
-    output << partial if partial
-  rescue IO::WaitReadable, EOFError
-  end
+messageable_was_published = false
 
-  break if output.include?("published message")
+published_messages = Outboxer::Message.list(status: :published)[:messages]
 
-  sleep delay
-  attempt += 1
+messageable_was_published = published_messages.any? do |published_message|
+    published_message[:messageable_type] == event.class.name &&
+    published_message[:messageable_id] == event.id.to_s
 end
 
-Process.kill("TERM", pid)
-Process.wait(pid)
-read_io.close
+while (attempt <= max_attempts) && !messageable_was_published
+    warn "Outboxer message not published yet. Retrying (#{attempt}/#{max_attempts})..."
+    sleep delay
+    attempt += 1
 
-if output.include?("published message")
-  puts "Outboxer published message found"
-  exit 0
-else
-  puts "Outboxer published message not found after #{max_attempts} attempts"
-  exit 1
+    published_messages = Outboxer::Message.list(status: :published)[:messages]
+
+    messageable_was_published = published_messages.any? do |published_message|
+        published_message[:messageable_type] == event.class.name &&
+        published_message[:messageable_id] == event.id.to_s
+    end
 end
+
+Process.kill("TERM", publisher_pid)
+Process.wait(publisher_pid)
+
+exit(messageable_was_published ? 0 : 1)
 RUBY
 
 # TARGET_RUBY_VERSION=3.2.2 TARGET_RAILS_VERSION=7.1.5.1 TARGET_DATABASE_ADAPTER=postgresql ./quickstart_e2e_tests.sh