feat: add redis and test sync adapter

Author: eliias

.gitignore

@@ -8,8 +8,9 @@
 /spec/dummy/log/*.log
 /spec/dummy/storage/
 /spec/dummy/tmp/
-Gemfile.lock
 .rspec_status
+dump.rdb
+Gemfile.lock
 
 # YARD docs
 doc/

examples/collaborative-text-editor/.gitignore

@@ -40,4 +40,4 @@
 /node_modules
 
 # Redis
-dump.rdb
+/dump.rdb

examples/collaborative-text-editor/config/sync.yml

@@ -0,0 +1,10 @@
+development:
+  adapter: redis
+  url: redis://localhost:6379/1
+
+test:
+  adapter: test
+
+production:
+  adapter: redis
+  url: <%= ENV.fetch("REDIS_URL") { "redis://localhost:6379/1" } %>

gems/yrb-actioncable/Gemfile

@@ -14,7 +14,7 @@ gem "y-rb"
 group :development, :test do
   gem "foreman"
   gem "jsbundling-rails"
-  gem "redis"
+  gem "redis", "~> 5.0.6"
   gem "rspec"
   gem "rspec-rails"
   gem "rubocop"

gems/yrb-actioncable/lib/y/actioncable.rb

@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
-require "y/actioncable/config"
+require "y/actioncable/adapter/redis"
+require "y/actioncable/adapter/test"
+require "y/actioncable/configuration"
 require "y/actioncable/engine"
 require "y/actioncable/reliable"
 require "y/actioncable/sync"
@@ -10,5 +12,8 @@
 module Y
   module Actioncable
     # Your code goes here...
+    module_function def config # rubocop:disable Style/AccessModifierDeclarations
+      @config ||= Y::Actioncable::Configuration.new
+    end
   end
 end

gems/yrb-actioncable/lib/y/actioncable/adapter/redis.rb

@@ -0,0 +1,109 @@
+# frozen_string_literal: true
+
+module Y
+  module Actioncable
+    module Adapter
+      # A distributed Y::Actioncable coordinator for reliable messaging.
+      class Redis
+        # @attr_reader [::Redis] client The redis client
+        attr_reader :client
+
+        # @param [Hash] config
+        def initialize(config)
+          @client = ::Redis.new(url: config[:url])
+        end
+
+        # @param [String] key
+        # @param [String] value
+        # @param [Numeric] offset
+        def add(key, value, offset = 0)
+          client.zadd(key, offset.to_f, value)
+
+          nil
+        end
+
+        # Append value to stream and return entry ID
+        #
+        # @param [String] key
+        # @param [Hash] value
+        # @return [String] The entry ID
+        def append(key, value)
+          client.xadd(key, value)
+        end
+
+        # @param [String] key
+        # @param [String] value
+        def remove(key, value)
+          client.zrem(key, value)
+
+          nil
+        end
+
+        # Move item by setting offset
+        #
+        # @param [String] key
+        # @param [String] value
+        # @param [Numeric] offset
+        def move(key, value, offset)
+          client.zadd(
+            key,
+            offset.to_f,
+            value,
+            gt: true
+          )
+
+          nil
+        end
+
+        # Return the minimum value in the stream identified by key
+        #
+        # @param [String] key
+        # @return [Numeric]
+        def min(key)
+          result = client.zrangebyscore(
+            key,
+            "-inf",
+            "+inf",
+            with_scores: true,
+            limit: [0, 1]
+          )
+
+          return 0 unless result
+          return 0 unless result.size.positive?
+
+          result.first[1].to_i
+        end
+
+        # Read values from stream starting (inclusive) from offset
+        #
+        # @param [String] key
+        # @param [String] offset
+        # @return [::Array<Object>]
+        def read(key, offset = nil)
+          offset ||= "-"
+
+          result = client.xrange(
+            key,
+            offset,
+            "+"
+          )
+
+          return [] unless result
+          return [] unless result.size.positive?
+
+          result
+        end
+
+        # Truncate the stream up until the given offset
+        #
+        # @param [String] key
+        # @param [String] offset
+        def truncate(key, offset)
+          client.xtrim(key, offset, strategy: "MINID")
+
+          nil
+        end
+      end
+    end
+  end
+end

gems/yrb-actioncable/lib/y/actioncable/adapter/test.rb

@@ -0,0 +1,131 @@
+# frozen_string_literal: true
+
+module Y
+  module Actioncable
+    module Adapter
+      # A local Y::Actioncable coordinator for reliable messaging.
+      class Test
+        # @attr_reader [Hash] store
+        attr_reader :store
+
+        def initialize
+          @store = {}
+          @last_ts = 0
+          @counter = 0
+        end
+
+        # @param [String] key
+        # @param [String] value
+        # @param [Numeric] offset
+        def add(key, value, offset = 0)
+          store[key] ||= {}
+          store[key][value] = offset
+
+          nil
+        end
+
+        # Append value to stream and return entry ID
+        #
+        # @param [String] key
+        # @param [Hash] value
+        # @return [String] The entry ID
+        def append(key, value)
+          curr_ts = Time.current.to_i
+          if curr_ts == @last_ts
+            @counter += 1
+          else
+            @counter = 0
+          end
+          @last_ts = curr_ts
+
+          entry_id = "#{Time.current.to_i}-#{@counter}"
+
+          store[key] ||= []
+          store[key] << [entry_id, value]
+
+          entry_id
+        end
+
+        # Return the minimum value in the stream identified by key
+        #
+        # @param [String] key
+        # @return [Numeric]
+        def min(key)
+          store[key] ||= {}
+          result = store[key].sort_by { |_k, v| -v }
+
+          return 0 unless result
+          return 0 unless result.size.positive?
+
+          result.first[1].to_i
+        end
+
+        # Move item by setting offset
+        #
+        # @param [String] key
+        # @param [String] value
+        # @param [Numeric] offset
+        def move(key, value, offset)
+          store[key] ||= {}
+          store[key][value] = offset if store[key][value] && offset > store[key][value]
+
+          nil
+        end
+
+        # Read values from stream starting (inclusive) from offset
+        #
+        # @param [String] key
+        # @param [String, nil] offset
+        # @return [::Array<Object>]
+        def read(key, offset = nil)
+          offset ||= "0-0"
+
+          store[key] ||= {}
+          store[key].filter do |(entry_id, _)|
+            ts1, counter1 = entry_id.split("-").map(&:to_i)
+            ts2, counter2 = offset.split("-").map(&:to_i)
+
+            gte = false
+            gte = true if ts1 >= ts2
+
+            gte = counter1 >= counter2 if ts1 == ts2
+
+            gte
+          end
+        end
+
+        # @param [String] key
+        # @param [String] value
+        def remove(key, value)
+          store[key] ||= {}
+          store[key].delete(value)
+
+          nil
+        end
+
+        # Truncate the stream up until the given offset (exclusive)
+        #
+        # [1, 2, 3].truncate(2) = [2, 3]
+        #
+        # @param [String] key
+        # @param [String] offset
+        def truncate(key, offset)
+          store[key] ||= {}
+          store[key].filter! do |(entry_id, _)|
+            ts1, counter1 = entry_id.split("-").map(&:to_i)
+            ts2, counter2 = offset.split("-").map(&:to_i)
+
+            gte = false
+            gte = true if ts1 >= ts2
+
+            gte = counter1 >= counter2 if ts1 == ts2
+
+            gte
+          end
+
+          nil
+        end
+      end
+    end
+  end
+end