Merge pull request #127 from cookpad/errm/outbox_message_bus

Add a message bus for transactional outbox deliveries

lib/streamy/event_handler.rb

@@ -1,4 +1,5 @@
 require "active_support/core_ext/hash/indifferent_access"
+require "ostruct"
 
 module Streamy
   class EventHandler

lib/streamy/message_buses/outbox_message_bus.rb

@@ -0,0 +1,22 @@
+require "streamy/kafka_configuration"
+require "waterdrop"
+require "active_support/core_ext/hash/indifferent_access"
+require "active_support/json"
+
+module Streamy
+  module MessageBuses
+    class OutboxMessageBus < MessageBus
+      def initialize(config)
+        @model = config[:model]
+      end
+
+      def deliver(key:, topic:, payload:, priority:)
+        @model.create(key: key, topic: topic, payload: payload)
+      end
+
+      def deliver_many(messages)
+        @model.create(messages.map { |message| message.except(:priority) })
+      end
+    end
+  end
+end

streamy.gemspec

@@ -45,4 +45,5 @@ Gem::Specification.new do |spec| # rubocop:disable Metrics/BlockLength
   spec.add_dependency "avro_turf", "~> 1.3.0"
   spec.add_dependency "waterdrop", ">= 2.4.10", "< 3.0.0"
   spec.add_dependency "webmock", "~> 3.3"
+  spec.add_dependency "ostruct"
 end

test/avro_deserializer_test.rb

@@ -1,6 +1,7 @@
 require "test_helper"
 require "avro_turf/test/fake_confluent_schema_registry_server"
 require "webmock/minitest"
+require "ostruct"
 
 module Streamy
   class AvroDeserializerTest < Minitest::Test

test/message_buses/outbox_message_bus_test.rb

@@ -0,0 +1,101 @@
+require "test_helper"
+require "waterdrop"
+require "streamy/message_buses/outbox_message_bus"
+
+module Streamy
+  class OutboxMessageBusTest < Minitest::Test
+    attr_reader :bus
+
+    def setup
+      @model = mock("outbox_model")
+      @bus = MessageBuses::OutboxMessageBus.new(model: @model)
+    end
+
+    def example_delivery(priority)
+      bus.deliver(
+        payload: payload.to_s,
+        key: "prk-sg-001",
+        topic: "charcuterie",
+        priority: priority
+      )
+    end
+
+    def payload
+      {
+        type: "sausage",
+        body: { meat: "pork", herbs: "sage" },
+        event_time: "2018"
+      }
+    end
+
+    def expected_event(key: "prk-sg-001")
+      {
+        payload: {
+          type: "sausage",
+          body: {
+            meat: "pork",
+            herbs: "sage"
+          },
+          event_time: "2018"
+        }.to_s,
+        key: key,
+        topic: "charcuterie"
+      }
+    end
+
+    def test_standard_priority_deliver
+      @model.expects(:create).with(expected_event)
+      example_delivery(:standard)
+    end
+
+    def test_low_priority_deliver
+      @model.expects(:create).with(expected_event)
+      example_delivery(:low)
+    end
+
+    def test_essential_priority_deliver
+      @model.expects(:create).with(expected_event)
+      example_delivery(:essential)
+    end
+
+    def test_all_priority_delivery
+      @model.expects(:create).with(expected_event)
+      example_delivery(:essential)
+
+      @model.expects(:create).with(expected_event)
+      example_delivery(:low)
+
+      @model.expects(:create).with(expected_event)
+      example_delivery(:standard)
+    end
+
+    def test_batch_delivery
+      @model.expects(:create).with([
+        expected_event(key: "prk-sg-001"),
+        expected_event(key: "prk-sg-002"),
+        expected_event(key: "prk-sg-003")
+      ])
+
+      bus.deliver_many([
+        {
+          payload: payload.to_s,
+          key: "prk-sg-001",
+          topic: "charcuterie",
+          priority: :standard
+        },
+        {
+          payload: payload.to_s,
+          key: "prk-sg-002",
+          topic: "charcuterie",
+          priority: :standard
+        },
+        {
+          payload: payload.to_s,
+          key: "prk-sg-003",
+          topic: "charcuterie",
+          priority: :standard
+        }
+      ])
+    end
+  end
+end