message-store and message-store-postgres are combined

Author: aaronjensen

lib/message_store.rb

@@ -1,5 +1,4 @@
-require 'pp'
-require 'json'
+require 'pg'
 
 require 'casing'
 require 'identifier/uuid'
@@ -9,9 +8,12 @@
 require 'virtual'
 require 'async_invocation'
 
+require 'log'
+require 'settings'
+
 require 'message_store/expected_version'
-require 'message_store/no_stream'
 require 'message_store/id'
+require 'message_store/no_stream'
 require 'message_store/stream_name'
 
 require 'message_store/message_data'
@@ -21,10 +23,20 @@
 
 require 'message_store/log'
 
+require 'message_store/settings'
+require 'message_store/session'
+
+require 'message_store/put'
+require 'message_store/write'
+
 require 'message_store/get'
 require 'message_store/get/substitute'
+require 'message_store/get/condition'
+require 'message_store/get/stream'
 require 'message_store/get/stream/last'
 require 'message_store/get/stream/last/substitute'
+require 'message_store/get/category'
+require 'message_store/get/category/correlation'
+require 'message_store/get/category/consumer_group'
 require 'message_store/read/iterator'
 require 'message_store/read'
-require 'message_store/write'

lib/message_store/controls.rb

@@ -8,12 +8,10 @@
 require 'message_store/controls/id'
 require 'message_store/controls/category'
 require 'message_store/controls/stream_name'
-require 'message_store/controls/read'
+require 'message_store/controls/position'
 require 'message_store/controls/message_data'
-require 'message_store/controls/message_data/hash'
 require 'message_store/controls/message_data/metadata'
-require 'message_store/controls/message_data/write'
+require 'message_store/controls/message_data/hash'
 require 'message_store/controls/message_data/read'
-require 'message_store/controls/write'
-require 'message_store/controls/get'
-require 'message_store/controls/get_last'
+require 'message_store/controls/message_data/write'
+require 'message_store/controls/put'

lib/message_store/controls/get_last.rb

@@ -48,6 +48,26 @@ def self.data
         def self.metadata
           MessageData::Metadata.data
         end
+
+        module List
+          Entry = Struct.new(:stream_name, :category, :message_data)
+
+          def self.get(instances: nil, stream_name: nil, category: nil)
+            instances ||= 1
+
+            list = []
+            instances.times do
+              instance_stream_name = stream_name || StreamName.example(category: category)
+              instance_category = MessageStore::StreamName.get_category(instance_stream_name)
+
+              write_message = Controls::MessageData::Write.example
+
+              list << Entry.new(instance_stream_name, instance_category, write_message)
+            end
+
+            list
+          end
+        end
       end
     end
   end

lib/message_store/controls/message_data/write.rb

@@ -0,0 +1,13 @@
+module MessageStore
+  module Controls
+    module Position
+      def self.example
+        1
+      end
+
+      def self.max
+        (2 ** 63) - 1
+      end
+    end
+  end
+end

lib/message_store/controls/position.rb

@@ -0,0 +1,26 @@
+module MessageStore
+  module Controls
+    module Put
+      def self.call(instances: nil, stream_name: nil, message_data: nil, message: nil, category: nil, type: nil)
+        instances ||= 1
+        stream_name ||= StreamName.example(category: category)
+        message_data ||= message
+
+        message_specified = !message_data.nil?
+
+        message_data ||= MessageData::Write.example(type: type)
+
+        position = nil
+        instances.times do
+          position = MessageStore::Put.(message_data, stream_name)
+
+          unless message_specified
+            message_data.id = MessageData::Write.id
+          end
+        end
+
+        [stream_name, position]
+      end
+    end
+  end
+end

lib/message_store/controls/put.rb

@@ -7,10 +7,161 @@ def self.included(cls)
         include Virtual
         include Log::Dependency
 
+        prepend Call
+        prepend BatchSize
+
+        dependency :session, Session
+
         abstract :call
         abstract :stream_name
-        abstract :batch_size
+        abstract :sql_command
+        abstract :parameters
+        abstract :parameter_values
         abstract :last_position
+        abstract :log_text
+
+        virtual :specialize_error
+        virtual :assure
+      end
+    end
+
+    module BatchSize
+      def batch_size
+        @batch_size ||= Defaults.batch_size
+      end
+    end
+
+    def self.build(stream_name, **args)
+      cls = specialization(stream_name)
+      cls.build(stream_name, **args)
+    end
+
+    def self.configure(receiver, stream_name, **args)
+      attr_name = args.delete(:attr_name)
+      attr_name ||= :get
+
+      instance = build(stream_name, **args)
+      receiver.public_send("#{attr_name}=", instance)
+    end
+
+    ## Is there a path to removing this? - Aaron, Sun Jan 15 2023
+    def configure(session: nil)
+      Session.configure(self, session: session)
+    end
+
+    def self.call(stream_name, **args)
+      position = args.delete(:position)
+      instance = build(stream_name, **args)
+      instance.(position)
+    end
+
+    module Call
+      def call(position=nil, stream_name: nil)
+        position ||= self.class::Defaults.position
+
+        stream_name ||= self.stream_name
+
+        assure
+
+        logger.trace(tag: :get) { "Getting message data (#{log_text(stream_name, position)})" }
+
+        result = get_result(stream_name, position)
+
+        message_data = convert(result)
+
+        logger.info(tag: :get) { "Finished getting message data (Count: #{message_data.length}, #{log_text(stream_name, position)})" }
+        logger.info(tags: [:data, :message_data]) { message_data.pretty_inspect }
+
+        message_data
+      end
+    end
+
+    def get_result(stream_name, position)
+      logger.trace(tag: :get) { "Getting result (#{log_text(stream_name, position)})" }
+
+      parameter_values = parameter_values(stream_name, position)
+
+      begin
+        result = session.execute(sql_command, parameter_values)
+      rescue PG::RaiseException => e
+        raise_error(e)
+      end
+
+      logger.debug(tag: :get) { "Finished getting result (Count: #{result.ntuples}, #{log_text(stream_name, position)})" }
+
+      result
+    end
+
+    def convert(result)
+      logger.trace(tag: :get) { "Converting result to message data (Result Count: #{result.ntuples})" }
+
+      message_data = result.map do |record|
+        Get.message_data(record)
+      end
+
+      logger.debug(tag: :get) { "Converted result to message data (Message Data Count: #{message_data.length})" }
+
+      message_data
+    end
+
+    def self.message_data(record)
+      record['data'] = Get::Deserialize.data(record['data'])
+      record['metadata'] = Get::Deserialize.metadata(record['metadata'])
+      record['time'] = Get::Time.utc_coerced(record['time'])
+
+      MessageData::Read.build(record)
+    end
+
+    def raise_error(pg_error)
+      error_message = Get.error_message(pg_error)
+
+      error = Condition.error(error_message)
+
+      if error.nil?
+        error = specialize_error(error_message)
+      end
+
+      if not error.nil?
+        logger.error { error_message }
+        raise error
+      end
+
+      raise pg_error
+    end
+
+    def self.error_message(pg_error)
+      pg_error.message.gsub('ERROR:', '').strip
+    end
+
+    def self.specialization(stream_name)
+      if StreamName.category?(stream_name)
+        Category
+      else
+        Stream
+      end
+    end
+
+    module Deserialize
+      def self.data(serialized_data)
+        return nil if serialized_data.nil?
+        Transform::Read.(serialized_data, :json, MessageData::Hash)
+      end
+
+      def self.metadata(serialized_metadata)
+        return nil if serialized_metadata.nil?
+        Transform::Read.(serialized_metadata, :json, MessageData::Hash)
+      end
+    end
+
+    module Time
+      def self.utc_coerced(local_time)
+        Clock::UTC.coerce(local_time)
+      end
+    end
+
+    module Defaults
+      def self.batch_size
+        1000
       end
     end
   end