Add Avro support

lib/streamy.rb

@@ -3,10 +3,13 @@ module Streamy
   # TODO: Move into classes that use them
   require "active_support"
   require "active_support/core_ext/string"
+  require 'byebug'
 
   require "streamy/version"
   require "streamy/consumer"
   require "streamy/event"
+  require "streamy/json_event"
+  require "streamy/avro_event"
   require "streamy/event_handler"
   require "streamy/message_processor"
   require "streamy/profiler"

lib/streamy/avro_event.rb

@@ -0,0 +1,17 @@
+require "active_support/core_ext/class/attribute"
+
+module Streamy
+  class AvroEvent < Event
+    def payload
+      avro.encode(payload_attributes.stringify_keys, schema_name: type)
+    end
+
+    def payload_attributes
+      {
+        type: type,
+        body: body,
+        event_time: event_time
+      }
+    end
+  end
+end

lib/streamy/event.rb

@@ -18,11 +18,9 @@ def self.publish(*args)
 
     def publish
       message_bus.safe_deliver(
+        payload: payload,
         key: key,
         topic: topic,
-        type: type,
-        body: body,
-        event_time: event_time,
         priority: priority
       )
     end

lib/streamy/json_event.rb

@@ -0,0 +1,13 @@
+require "active_support/core_ext/class/attribute"
+
+module Streamy
+  class JsonEvent < Event
+    def payload
+      {
+        type: type,
+        body: body,
+        event_time: event_time
+      }.to_json
+    end
+  end
+end

lib/streamy/message_buses/kafka_message_bus.rb

@@ -10,13 +10,7 @@ def initialize(config)
         @kafka = Kafka.new(kafka_config)
       end
 
-      def deliver(key:, topic:, type:, body:, event_time:, priority:)
-        payload = {
-          type: type,
-          body: body,
-          event_time: event_time
-        }.to_json
-
+      def deliver(key:, topic:, payload:, priority:)
         producer(priority).tap do |p|
           p.produce(payload, key: key, topic: topic)
           case priority

lib/streamy/message_buses/message_bus.rb

@@ -7,7 +7,7 @@ def safe_deliver(*args)
         raise PublicationFailedError.new(e, *args)
       end
 
-      def deliver(key:, topic:, type:, body:, event_time:, priority:)
+      def deliver(key:, topic:, payload:, priority:)
         raise "not implemented"
       end
     end

streamy.gemspec

@@ -32,6 +32,7 @@ Gem::Specification.new do |spec| # rubocop:disable Metrics/BlockLength
   spec.require_paths = ["lib"]
 
   spec.add_development_dependency "bundler", "~> 1.14"
+  spec.add_development_dependency "byebug"
   spec.add_development_dependency "minitest", "~> 5.0"
   spec.add_development_dependency "minitest-focus"
   spec.add_development_dependency "mocha", "~> 1.4"

test/event_test.rb → test/json_event_test.rb

@@ -1,44 +1,44 @@
 require "test_helper"
 
 module Streamy
-  class EventTest < Minitest::Test
-    class EventWithoutTopic < Event
+  class JsonEventTest < Minitest::Test
+    class EventWithoutTopic < JsonEvent
       def event_time; end
 
       def body; end
     end
 
-    class EventWithoutEventTime < Event
+    class EventWithoutEventTime < JsonEvent
       def topic; end
 
       def body; end
     end
 
-    class EventWithoutBody < Event
+    class EventWithoutBody < JsonEvent
       def topic; end
 
       def event_time; end
     end
 
     def test_helpful_error_message_on_missing_topic
-      assert_runtime_error "topic must be implemented on Streamy::EventTest::EventWithoutTopic" do
+      assert_runtime_error "topic must be implemented on Streamy::JsonEventTest::EventWithoutTopic" do
         EventWithoutTopic.publish
       end
     end
 
     def test_helpful_error_message_on_missing_event_time
-      assert_runtime_error "event_time must be implemented on Streamy::EventTest::EventWithoutEventTime" do
+      assert_runtime_error "event_time must be implemented on Streamy::JsonEventTest::EventWithoutEventTime" do
         EventWithoutEventTime.publish
       end
     end
 
     def test_helpful_error_message_on_missing_body
-      assert_runtime_error "body must be implemented on Streamy::EventTest::EventWithoutBody" do
+      assert_runtime_error "body must be implemented on Streamy::JsonEventTest::EventWithoutBody" do
         EventWithoutBody.publish
       end
     end
 
-    class TestEvent < Event
+    class TestEvent < JsonEvent
       def topic
         :bacon
       end

test/message_buses/kafka_message_bus_test.rb

@@ -21,15 +21,21 @@ def teardown
 
     def example_delivery(priority)
       bus.deliver(
+        payload: payload,
         key: "prk-sg-001",
-        type: "sausage",
         topic: "charcuterie",
-        event_time: "2018",
-        body: { meat: "pork", herbs: "sage" },
         priority: priority
       )
     end
 
+    def payload
+      {
+        type: "sausage",
+        body: { meat: "pork", herbs: "sage" },
+        event_time: "2018"
+      }.to_json
+    end
+
     def expected_event
       [
         {