Add exteranl payload store

.circleci/config.yml

@@ -0,0 +1,69 @@
+defaults: &defaults
+  working_directory: ~/mantle
+  parallelism: 1
+  docker:
+      - image: circleci/ruby:2.6.2
+        environment:
+          RAILS_ENV: test
+          PGHOST: 127.0.0.1
+          PGUSER: postgres
+      - image: circleci/postgres:9.3-alpine
+        environment:
+          POSTGRES_USER: postgres
+      - image: redis
+
+version: 2
+jobs:
+  build:
+    <<: *defaults
+    steps:
+      - run: sudo apt-get update && sudo apt-get install -y r-base postgresql-client || true
+
+      - restore_cache:
+          key: v1-mantle-repo-{{ .Environment.CIRCLE_SHA1 }}
+
+      - checkout
+
+      - save_cache:
+          key: v1-mantle-repo-{{ .Environment.CIRCLE_SHA1 }}
+          paths:
+            - ~/mantle
+
+      - restore_cache:
+          key: v1-mantle-bundle-{{ checksum "Gemfile.lock" }}
+
+      - run:
+          name: install bundler
+          command: |
+            gem install bundler:2.1.4
+
+      - run:
+          name: bundle install
+          command: |
+            bundle install --jobs=4 --retry=3 --path vendor/bundle
+
+      - save_cache:
+          paths:
+            - ~/mantle/vendor/bundle
+          key: v1-mantle-bundle-{{ checksum "Gemfile.lock" }}
+
+      - run:
+          name: Wait for DB
+          command: dockerize -wait tcp://localhost:5432 -timeout 1m
+
+      - run:
+          name: Rspec
+          command: |
+            mkdir /tmp/test-results
+            TEST_FILES="$(circleci tests glob "spec/**/*_spec.rb" | circleci tests split --split-by=timings)"
+
+            bundle exec rspec --format progress \
+                              --out test_results/rspec.xml \
+                              -- $TEST_FILES
+
+      - store_test_results:
+          path: test_results
+
+      - store_artifacts:
+          path: test-results/rspec.xml
+          destination: test-results

Gemfile.lock

@@ -1,22 +1,23 @@
 PATH
   remote: .
   specs:
-    mantle (2.3.0)
+    mantle (2.4.0)
       redis
       sidekiq (~> 5.0)
+      uuidtools
 
 GEM
   remote: https://rubygems.org/
   specs:
     coderay (1.1.0)
-    connection_pool (2.2.2)
+    connection_pool (2.2.3)
     diff-lcs (1.2.5)
     method_source (0.8.2)
     pry (0.10.0)
       coderay (~> 1.1.0)
       method_source (~> 0.8.1)
       slop (~> 3.4)
-    rack (2.0.9)
+    rack (2.2.3)
     rack-protection (2.0.8.1)
       rack
     redis (4.1.3)
@@ -33,12 +34,13 @@ GEM
       diff-lcs (>= 1.2.0, < 2.0)
       rspec-support (~> 3.2.0)
     rspec-support (3.2.2)
-    sidekiq (5.2.8)
+    sidekiq (5.2.9)
       connection_pool (~> 2.2, >= 2.2.2)
-      rack (< 2.1.0)
+      rack (~> 2.0)
       rack-protection (>= 1.5.0)
-      redis (>= 3.3.5, < 5)
+      redis (>= 3.3.5, < 4.2)
     slop (3.5.0)
+    uuidtools (2.2.0)
 
 PLATFORMS
   ruby
@@ -49,4 +51,4 @@ DEPENDENCIES
   rspec
 
 BUNDLED WITH
-   2.1.4
+   2.3.5

README.md

@@ -19,6 +19,8 @@ or install manually by:
 
 ## Usage (in Rails App)
 
+### Configure
+
 Setup a Rails initializer(`config/initializers/mantle.rb`):
 
 
@@ -60,20 +62,40 @@ Mantle.configure do |config|
 end
 ```
 
+### Publish Messages (Publisher)
 
 Publish messages to consumers:
 
 ```Ruby
-Mantle::Message.new("person:create").publish({ id: message['id'], data: message['data'] })
+Mantle::Message.new("person:create").publish(message: { id: message['id'], data: message['data'] })
 ```
 
 The first and only argument to `Mantle::Message.new` is the channel you want to publish the
-message on. The `#publish` method takes the message payload (in any format you like)
-and pushes the message on to the message bus pub/sub and also adds it to the
+message on. The `#publish` method takes a named argument `message:` which contains the message content (in any structure you like).
+This pushes the `message` on to the message bus pub/sub and also adds it to the
 catch up queue so offline applications can process the message when they become available.
 
+Note that you can still use a bare argument for the message, but this will be deprecated in the future:
+
+```Ruby
+Mantle::Message.new("person:create").publish({ id: message['id'], data: message['data'] })
+```
+
+### Receive Messages (Consumer)
+
 Define message handler class with `.receive` method. For example `app/models/my_message_handler.rb`
 
+```Ruby
+class MyMessageHandler
+  def self.receive(channel:, message:)
+    puts channel # => 'order'
+    puts message # => { 'id' => 5, 'name' => 'Brandon' }
+  end
+end
+```
+
+Note that you can still use two bare arguments for the channel and message, but this will be deprecated in the future:
+
 ```Ruby
 class MyMessageHandler
   def self.receive(channel, message)
@@ -83,6 +105,9 @@ class MyMessageHandler
 end
 ```
 
+
+### Listener / Processor
+
 To run the listener:
 
 ```
@@ -111,6 +136,140 @@ $ bin/sidekiq -q mantle -q default
 It will NOT add the `default` queue to processing if there are other queues
 enumerated using the `-q` option.
 
+### Large Payloads
+
+Because Mantle uses Redis for the message bus, sending very large messages can quickly use a lot of Redis store,
+and in the event that Redis memory is exceeded, your app will be rendered inoperable. In addition, a Mantle handle may
+need to pass the `message` on to other services (for example, queue processors), passing very large messages can
+compound resulting in even greater memory usage for the same exact payload.
+
+For this reason, it sometimes makes sense to send large payloads through an external key/value store where the handler can
+retrieve the payload only when needed instead of pass the entire payload as part of the message.
+
+Note that rather than move the entire payload to external store, it often makes sense for the handler to have a small amount
+of data available without retrieving the external payload, such as `account_id` so the handler can do something like this in
+the top of the handler (this reveals a named argument `uuid` which is documented below):
+
+```Ruby
+class MyMessageHandler
+  def self.receive(channel:, message:, uuid:)
+    return unless interesting_account?(message['account_id'])
+
+    puts channel # => 'order'
+    puts message # => { 'id' => 5, 'name' => 'Brandon' }
+  end
+end
+```
+
+#### Configuring External Store
+
+Mantle facilitates sending large payloads with limited impact on your message bus memory usage by allowing you to
+configure an external store by adding the following in the initializer.
+
+External store can use either `redis` (optionally, a different `Redis` instance) or `ActiveRecord`.
+
+To use `redis` use a hash to configure an `external_store_manager`:
+
+``` Ruby
+Mantle.configure do |config|
+  ...
+  config.external_store_manager = { redis: Redis.new(host: 'localhost'), keep_for: 3.hours } # default: keep_for: nil
+  ...
+end
+```
+
+To use `ActiveRecord` use a hash to configure an `external_store_manager`:
+
+``` Ruby
+Mantle.configure do |config|
+  ...
+  config.external_store_manager = { table_name: `my_external_payloads`, database: {...}, keep_for: 3.hours } # default: keep_for: nil
+  ...
+end
+```
+
+The `database` hash will be passed to ActiveRecord `establish_connection`.
+
+The `table_name` specified must be a table in the database, and must be creating using the following migration:
+
+```Ruby
+  create_table :my_external_payloads do |t|
+    t.string :uuid, nil: false
+    t.text :payload, nil: false
+    t.timestamp :keep_until, nil: true
+    t.timestamp :expire_at, nil: true
+    t.timestamp :created_at, nil: false
+  end
+
+  add_index :my_external_payloads, :uuid
+  add_index :my_external_payloads, :expire_at
+  add_index :my_external_payloads, :created_at
+```
+
+#### Publishing with External Payloads
+
+An external payload can added to the publish method as using a named argument, `payload:`:
+
+```Ruby
+the_payload = { body: 'large_external_payload' }
+the_message = { id: message['id'], data: message['data'] }
+Mantle::Message.new("person:create").publish(payload: the_payload, message: the_message)
+```
+
+There are three ways the `ExternalStoreManager` will free memory.
+- If a `keep_until` parameter is specified, then the payload will not be freed until that time. The payload may outlive the specified time, based on `least recently created`.
+- If an `expire_at` parameter is specified, then the payload will be freed at that time. The payload will not outlive the specified time.
+- If neither qualifier is specified by the publisher, then `least recently created` will be freed, as needed.
+
+```Ruby
+Mantle::Message.new("person:create").publish(message: { id: message['id'], data: message['data'] }, payload: { body: 'large_external_payload' }, keep_until: 3.hours.from_now)
+
+Mantle::Message.new("person:create").publish(message: { id: message['id'], data: message['data'] }, payload: { body: 'large_external_payload' }, expire_at: 3.hours.from_now)
+```
+
+#### Retrieving External Payloads
+
+A handler (consumer) does not need to be aware there is an external payload. If it does not define a named argument `uuid`,
+then the Mantle processor will retrieve the payload and merge it into the message before calling the handler.
+
+If, however, the handler is aware of the extneral payload, then it simply defineds a named argument `uuid` in the method, and
+the `uuid` will be set, and the `payload` will not be merged into message.
+
+```Ruby
+class MyMessageHandler
+  def self.receive(channel:, message:, uuid:)
+    puts channel # => 'order'
+    puts message # => { 'id' => 5, 'name' => 'Brandon' }
+    puts external_store_uuid # => ''
+    puts Mantle.retrieve_external_payload(uuid) # => { 'body' => 'large_external_payload' }
+  end
+end
+```
+
+One may want to keep the payload separate from the message so that the handler can pass the `uuid` as a parameter to a queue processor. This would avoid
+always adding large payloads to Sidekiq parameters (for example). In this case, the sidekiq processor would also need to be aware of the `uuid` and can call
+
+```Reuby
+   Mantle.external_store_managers.retrieve(uuid: uuid)
+```
+
+#### Using External Store Directly
+
+Using the concept of avoiding sending large payloads to queue processors may make senese even outside of Mantle handlers.
+
+For this reason, the `external_store_manager` is available to be used outside of Mantle:
+
+```Reuby
+   uuid = Mantle.external_store_managers.store(payload: "my large payload")
+```
+
+and then within the processor:
+
+
+```Reuby
+   Mantle.external_store_managers.retrieve(uuid: uuid)
+```
+
 ## Testing
 
 Requiring this library causes messages to be appended to an in-memory array.

lib/mantle.rb

@@ -2,6 +2,7 @@
 require 'redis'
 require 'sidekiq'
 require 'json'
+require 'uuidtools'
 
 begin
   require 'pry'
@@ -11,6 +12,9 @@
 require_relative 'mantle/catch_up'
 require_relative 'mantle/configuration'
 require_relative 'mantle/error'
+require_relative 'mantle/external_store_manager'
+require_relative 'mantle/external_store/redis'
+require_relative 'mantle/external_store/active_record'
 require_relative 'mantle/local_redis'
 require_relative 'mantle/logger'
 require_relative 'mantle/message'
@@ -42,6 +46,10 @@ def self.receive_message(channel, message)
     self.configuration.message_handlers.receive_message channel, message
   end
 
+  def self.external_store_manager
+    configuration.external_store_manager
+  end
+
   def self.channels
     configuration.message_handlers.channels
   end

lib/mantle/configuration.rb

@@ -5,7 +5,8 @@ class Configuration
                   :redis_namespace,
                   :whoami
 
-    attr_reader :message_handlers
+    attr_reader   :message_handlers,
+                  :external_store_manager
 
     def initialize
       @message_handlers = Mantle::MessageHandlers.new
@@ -16,5 +17,11 @@ def initialize
     def message_handlers=(hash_instance)
       @message_handlers = Mantle::MessageHandlers.new(hash_instance)
     end
+
+    def external_store=(args)
+      external_store, options = args
+      @external_store_manager ||= Mantle::ExternalStoreManager.new
+      @external_store_manager.configure(external_store, options || {})
+    end
   end
 end

lib/mantle/external_store/active_record.rb

@@ -0,0 +1,18 @@
+module Mantle
+  module ExternalStore
+    class ActiveRecord
+      def configure(options)
+        @table = options[:table]
+      end
+
+      def store(external_payload)
+        # TODO: implement actual store for active_record
+        'uuid'
+      end
+
+      def retriev(uuid)
+        # TODO: implement actual retrieve for active_record
+      end
+    end
+  end
+end