Postgres

.gitignore

@@ -1,8 +1,10 @@
 .bundle/
 Gemfile.lock
+*.log
 *.gem
 gems
 *sketch*
 *notes.*
 todo.md
 fixtures
+/test/interactive/stream_name.tmp

README.md

@@ -1,13 +1,11 @@
 # messaging
 
-Common primitives for platform-specific messaging implementations for Eventide.
-
-This library provides generalizations that are used by platform-specific implementations that provide concrete functionality.
+Eventide messaging for Postgres
 
 ## Documentation
 
 See the [Eventide documentation site](http://docs.eventide-project.org) for more information, examples, and user guides.
 
 ## License
 
-The `messaging` library is released under the [MIT License](https://github.com/eventide-project/messaging/blob/master/MIT-License.txt).
+The `messaging` library is released under the [MIT License](https://github.com/eventide-project/messaging-postgres/blob/master/MIT-License.txt).

init.rb

@@ -1,3 +1,3 @@
-require_relative './load_path'
+require_relative 'load_path'
 
 require 'messaging'

lib/messaging/controls.rb

@@ -13,6 +13,5 @@
 require 'messaging/controls/message_data'
 require 'messaging/controls/message'
 require 'messaging/controls/batch'
-require 'messaging/controls/write'
 require 'messaging/controls/settings'
 require 'messaging/controls/handler'

lib/messaging/write.rb

@@ -1,48 +1,30 @@
 module Messaging
-  module Write
-    class Error < RuntimeError; end
-
-    def self.included(cls)
-      cls.class_exec do
-        include Dependency
-        include Virtual
-        include Log::Dependency
-
-        dependency :message_writer
-        dependency :telemetry, ::Telemetry
+  class Write
+    include Dependency
+    include Virtual
+    include Log::Dependency
 
-        extend Build
-        extend Call
-        extend Configure
-
-        abstract :configure
+    class Error < RuntimeError; end
 
-        const_set :Substitute, Substitute
-      end
-    end
+    dependency :message_writer, MessageStore::Write
+    dependency :telemetry, ::Telemetry
 
-    module Build
-      def build(session: nil)
-        instance = new
-        instance.configure(session: session)
-        ::Telemetry.configure instance
-        instance
-      end
+    def self.build(session: nil)
+      instance = new
+      MessageStore::Write.configure(instance, attr_name: :message_writer, session: session)
+      ::Telemetry.configure(instance)
+      instance
     end
 
-    module Configure
-      def configure(receiver, session: nil, attr_name: nil)
-        attr_name ||= :write
-        instance = build(session: session)
-        receiver.public_send "#{attr_name}=", instance
-      end
+    def self.configure(receiver, session: nil, attr_name: nil)
+      attr_name ||= :write
+      instance = build(session: session)
+      receiver.public_send "#{attr_name}=", instance
     end
 
-    module Call
-      def call(message, stream_name, expected_version: nil, reply_stream_name: nil, session: nil)
-        instance = build(session: session)
-        instance.(message, stream_name, expected_version: expected_version, reply_stream_name: reply_stream_name)
-      end
+    def self.call(message, stream_name, expected_version: nil, reply_stream_name: nil, session: nil)
+      instance = build(session: session)
+      instance.(message, stream_name, expected_version: expected_version, reply_stream_name: reply_stream_name)
     end
 
     def call(message_or_batch, stream_name, expected_version: nil, reply_stream_name: nil)

lib/messaging/write/substitute.rb

@@ -1,5 +1,5 @@
 module Messaging
-  module Write
+  class Write
     module Substitute
       def self.build
         Substitute::Write.build.tap do |substitute_writer|
@@ -10,9 +10,7 @@ def self.build
 
       Error = Class.new(RuntimeError)
 
-      class Write
-        include Messaging::Write
-
+      class Write < Messaging::Write
         attr_accessor :sink
 
         def raise_expected_version_error
@@ -32,7 +30,6 @@ def call(*args, **keyword_args)
         end
         alias :write :call
 
-        ## Should never have had an exclamation mark - Scott, Tue Jan 17 2023
         def raise_expected_version_error!
           self.raise_expected_version_error = true
           nil

messaging.gemspec

@@ -2,7 +2,7 @@
 Gem::Specification.new do |s|
   s.name = 'evt-messaging'
   s.version = '2.7.0.1'
-  s.summary = 'Common primitives for platform-specific messaging implementations for Eventide'
+  s.summary = 'Eventide messaging for Postgres'
   s.description = ' '
 
   s.authors = ['The Eventide Project']
@@ -15,6 +15,9 @@ Gem::Specification.new do |s|
   s.platform = Gem::Platform::RUBY
   s.required_ruby_version = '>= 2.4.0'
 
+  ## Remove after message store is published w/ pg dependency
+  s.add_runtime_dependency 'pg'
+
   s.add_runtime_dependency 'evt-message_store'
   s.add_runtime_dependency 'evt-settings'
 

settings/message_store_postgres.json

@@ -0,0 +1,14 @@
+{
+  "dbname": "message_store",
+  "host": "localhost",
+  "hostaddr": "127.0.0.1",
+  "port": 5432,
+  "user": "message_store",
+  "password": null,
+  "connect_timeout": null,
+  "options": null,
+  "sslmode": null,
+  "krbsrvname": null,
+  "gsslib": null,
+  "service": null
+}

test/automated/write/batch/no_id.rb

@@ -0,0 +1,29 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Batch" do
+    context "No Message IDs" do
+      stream_name = Controls::StreamName.example
+
+      batch, values = Controls::Batch.example(id: :none)
+
+      batch.each do |message|
+        assert(message.id.nil?)
+      end
+
+      position = Write.(batch, stream_name)
+
+      context "Individual Events are Written" do
+        2.times do |i|
+          read_message = MessageStore::Get.(stream_name, position: i, batch_size: 1).first
+
+          context "Assigns an ID" do
+            test "Event #{i + 1}" do
+              refute(read_message.id.nil?)
+            end
+          end
+        end
+      end
+    end
+  end
+end

test/automated/write/batch/reply.rb

@@ -10,7 +10,7 @@
 
       batch = [message_1, message_2]
 
-      write = Controls::Write.example
+      write = Write.new
 
       test "Is an error" do
         assert_raises(Messaging::Write::Error) do

test/automated/write/batch/reply_is_an_error.rb

@@ -0,0 +1,3 @@
+require_relative '../../automated_init'
+
+comment "Tested in Messaging generalization library"

test/automated/write/batch/write.rb

@@ -0,0 +1,25 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Batch" do
+    stream_name = Controls::StreamName.example
+
+    batch, values = Controls::Batch.example
+
+    last_written_position = Write.(batch, stream_name)
+
+    test "Last written position" do
+      assert(last_written_position == 1)
+    end
+
+    context "Individual Events are Written" do
+      2.times do |i|
+        read_message = MessageStore::Get.(stream_name, position: i, batch_size: 1).first
+
+        test "Event #{i + 1}" do
+          assert(read_message.data[:some_attribute] == values[i])
+        end
+      end
+    end
+  end
+end

test/automated/write/batch/write_initial.rb

@@ -0,0 +1,42 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Batch" do
+    context "Writing the initial message to a stream that has not been created yet" do
+      stream_name = Controls::StreamName.example
+
+      batch, values = Controls::Batch.example
+
+      write = Write.build
+
+      write.initial(batch, stream_name)
+
+      context "Individual Events are Written" do
+        2.times do |i|
+          read_message = MessageStore::Get.(stream_name, position: i, batch_size: 1).first
+
+          test "Event #{i + 1}" do
+            assert(read_message.data[:some_attribute] == values[i])
+          end
+        end
+      end
+    end
+
+    context "Writing the initial message to a stream that already exists" do
+      stream_name = Controls::StreamName.example
+
+      batch = Controls::Batch::Messages.example
+
+      write = Write.build
+
+      message = Controls::Message.example
+      Write.(message, stream_name)
+
+      test "Is an error" do
+        assert_raises(MessageStore::ExpectedVersion::Error) do
+          write.initial(batch, stream_name)
+        end
+      end
+    end
+  end
+end

test/automated/write/batch/write_with_reply_stream.rb

@@ -0,0 +1,24 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Batch" do
+    context "With Reply Stream" do
+      stream_name = Controls::StreamName.example
+      reply_stream_name = Controls::StreamName.example
+
+      batch, values = Controls::Batch.example
+
+      Write.(batch, stream_name, reply_stream_name: reply_stream_name)
+
+      context "Individual Events are Written" do
+        2.times do |i|
+          read_message = MessageStore::Get.(stream_name, position: i, batch_size: 1).first
+
+          test "Event #{i + 1}" do
+            assert(read_message.data[:some_attribute] == values[i])
+          end
+        end
+      end
+    end
+  end
+end

test/automated/write/expected_version.rb

@@ -0,0 +1,29 @@
+require_relative '../automated_init'
+
+context "Write" do
+  context "Single Message" do
+    stream_name = Controls::StreamName.example(category: 'testWrongVersion')
+
+    message_1 = Controls::Message.example
+
+    Write.(message_1, stream_name)
+
+    message_2 = Controls::Message.example
+
+    context "Right Version" do
+      test "Succeeds" do
+        refute_raises(MessageStore::ExpectedVersion::Error) do
+          Write.(message_2, stream_name, expected_version: 0)
+        end
+      end
+    end
+
+    context "Wrong Version" do
+      test "Fails" do
+        assert_raises(MessageStore::ExpectedVersion::Error) do
+          Write.(message_2, stream_name, expected_version: 11)
+        end
+      end
+    end
+  end
+end

test/automated/write/message/no_id.rb

@@ -0,0 +1,21 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Message" do
+    context "No ID" do
+      stream_name = Controls::StreamName.example
+
+      message = Controls::Message.example(id: :none)
+
+      assert(message.id.nil?)
+
+      position = Write.(message, stream_name)
+
+      read_message = MessageStore::Get.(stream_name, position: position, batch_size: 1).first
+
+      test "Assigns an ID" do
+        refute(read_message.id.nil?)
+      end
+    end
+  end
+end

test/automated/write/message/reply.rb

@@ -0,0 +1,26 @@
+require_relative '../../automated_init'
+
+context "Write" do
+  context "Reply" do
+    context "Message" do
+      message = Controls::Message.example
+
+      reply_stream_name = Controls::StreamName.example
+      message.metadata.reply_stream_name = reply_stream_name
+
+      write = Write.build
+
+      position = write.reply(message)
+
+      read_message = MessageStore::Get.(reply_stream_name, position: position, batch_size: 1).first
+
+      test "Writes the message to the reply stream" do
+        assert(read_message.data == message.to_h)
+      end
+
+      test "Clears the reply stream from the metadata" do
+        assert(read_message.metadata[:reply_stream_name].nil?)
+      end
+    end
+  end
+end