FEATURE: support custom codecs for transport

Gemfile

@@ -18,6 +18,7 @@ end
 
 group :test, :development do
   gem 'byebug'
+  gem 'oj'
 end
 
 group :development do

lib/message_bus.rb

@@ -2,18 +2,22 @@
 
 require "monitor"
 require "set"
-require "message_bus/version"
-require "message_bus/message"
-require "message_bus/client"
-require "message_bus/connection_manager"
-require "message_bus/diagnostics"
-require "message_bus/rack/middleware"
-require "message_bus/rack/diagnostics"
-require "message_bus/timer_thread"
+
+require_relative "message_bus/version"
+require_relative "message_bus/message"
+require_relative "message_bus/client"
+require_relative "message_bus/connection_manager"
+require_relative "message_bus/diagnostics"
+require_relative "message_bus/rack/middleware"
+require_relative "message_bus/rack/diagnostics"
+require_relative "message_bus/timer_thread"
+require_relative "message_bus/codec/base"
+require_relative "message_bus/backends"
+require_relative "message_bus/backends/base"
 
 # we still need to take care of the logger
 if defined?(::Rails)
-  require 'message_bus/rails/railtie'
+  require_relative 'message_bus/rails/railtie'
 end
 
 # @see MessageBus::Implementation
@@ -278,6 +282,17 @@ def allow_broadcast?
       end
   end
 
+  # @param [MessageBus::Codec::Base] codec used to encode and decode Message payloads
+  # @return [void]
+  def transport_codec=(codec)
+    configure(trasport_codec: codec)
+  end
+
+  # @return [MessageBus::Codec::Base] codec used to encode and decode Message payloads
+  def transport_codec
+    @config[:transport_codec] ||= MessageBus::Codec::Json.new
+  end
+
   # @param [MessageBus::Backend::Base] pub_sub a configured backend
   # @return [void]
   def reliable_pub_sub=(pub_sub)
@@ -358,7 +373,7 @@ def publish(channel, data, opts = nil)
       raise ::MessageBus::InvalidMessageTarget
     end
 
-    encoded_data = JSON.dump(
+    encoded_data = transport_codec.encode(
       data: data,
       user_ids: user_ids,
       group_ids: group_ids,
@@ -626,7 +641,7 @@ def decode_message!(msg)
     channel, site_id = decode_channel_name(msg.channel)
     msg.channel = channel
     msg.site_id = site_id
-    parsed = JSON.parse(msg.data)
+    parsed = transport_codec.decode(msg.data)
     msg.data = parsed["data"]
     msg.user_ids = parsed["user_ids"]
     msg.group_ids = parsed["group_ids"]

lib/message_bus/backends/base.rb

@@ -1,7 +1,5 @@
 # frozen_string_literal: true
 
-require "message_bus/backends"
-
 module MessageBus
   module Backends
     # Backends provide a consistent API over a variety of options for persisting

lib/message_bus/backends/memory.rb

@@ -1,7 +1,5 @@
 # frozen_string_literal: true
 
-require "message_bus/backends/base"
-
 module MessageBus
   module Backends
     # The memory backend stores published messages in a simple array per

lib/message_bus/backends/postgres.rb

@@ -2,8 +2,6 @@
 
 require 'pg'
 
-require "message_bus/backends/base"
-
 module MessageBus
   module Backends
     # The Postgres backend stores published messages in a single Postgres table

lib/message_bus/backends/redis.rb

@@ -3,8 +3,6 @@
 require 'redis'
 require 'digest'
 
-require "message_bus/backends/base"
-
 module MessageBus
   module Backends
     # The Redis backend stores published messages in Redis sorted sets (using

lib/message_bus/codec/base.rb

@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+module MessageBus
+  module Codec
+    class Base
+      def encode(data:, user_ids:, group_ids:, client_ids:)
+        raise ConcreteClassMustImplementError
+      end
+
+      def decode(payload)
+        raise ConcreteClassMustImplementError
+      end
+    end
+
+    autoload :Json, File.expand_path("json", __dir__)
+    autoload :Oj, File.expand_path("oj", __dir__)
+    autoload :OjFast, File.expand_path("oj_fast", __dir__)
+  end
+end

lib/message_bus/codec/json.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module MessageBus
+  module Codec
+    class Json < Base
+      def encode(data:, user_ids:, group_ids:, client_ids:)
+        JSON.dump(
+          data: data,
+          user_ids: user_ids,
+          group_ids: group_ids,
+          client_ids: client_ids
+        )
+      end
+
+      def decode(payload)
+        JSON.parse(payload)
+      end
+    end
+  end
+end

lib/message_bus/codec/oj.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+require 'oj'
+
+module MessageBus
+  module Codec
+    class Oj < Base
+      def encode(data:, user_ids:, group_ids:, client_ids:)
+        ::Oj.dump({
+            data: data,
+            user_ids: user_ids,
+            group_ids: group_ids,
+            client_ids: client_ids
+          },
+          mode: :compat)
+      end
+
+      def decode(payload)
+        ::Oj.load(payload, mode: :compat)
+      end
+    end
+  end
+end

lib/message_bus/codec/oj_fast.rb

@@ -0,0 +1,81 @@
+# frozen_string_literal: true
+
+require 'oj'
+
+module MessageBus
+  module Codec
+    class FastIdList
+      def self.from_array(array)
+        new(array.sort.pack("V*"))
+      end
+
+      def self.from_string(string)
+        new(string)
+      end
+
+      def initialize(packed)
+        @packed = packed
+      end
+
+      def include?(id)
+        found = (0...length).bsearch do |index|
+          start = index * 4
+          @packed[start, start + 4].unpack1("V") >= id
+        end
+
+        if found
+          start = found * 4
+          found && @packed[start, start + 4].unpack1("V") == id
+        end
+      end
+
+      def length
+        @length ||= @packed.bytesize / 4
+      end
+
+      def to_a
+        @packed.unpack("V*")
+      end
+
+      def to_s
+        @packed
+      end
+    end
+
+    class OjFast < Base
+      def encode(data:, user_ids:, group_ids:, client_ids:)
+
+        if user_ids
+          user_ids = FastIdList.from_array(user_ids).to_s
+        end
+
+        #if group_ids
+        #  group_ids = FastIdList.from_array(group_ids).to_s
+        #end
+
+        ::Oj.dump({
+            data: data,
+            user_ids: user_ids,
+            group_ids: group_ids,
+            client_ids: client_ids
+          },
+          mode: :compat)
+      end
+
+      def decode(payload)
+        result = ::Oj.load(payload, mode: :compat)
+
+        if str = result["user_ids"]
+          result["user_ids"] = FastIdList.from_string(str)
+        end
+
+        # groups need to implement (-)
+        # if str = result["group_ids"]
+        #  result["group_ids"] = FastIdList.from_string(str)
+        # end
+
+        result
+      end
+    end
+  end
+end

spec/lib/message_bus_spec.rb

@@ -162,7 +162,7 @@
     data.must_equal 'norris'
     site_id.must_equal 'magic'
     channel.must_equal '/chuck'
-    user_ids.must_equal [1, 2, 3]
+    user_ids.to_a.must_equal [1, 2, 3]
   end
 
   it "should get global messages if it subscribes to them" do