THREESCALE-8404: Add ACL and TLS support for Redis

lib/3scale/backend/storage_async.rb

@@ -1,3 +1,4 @@
+require '3scale/backend/storage_async/methods'
 require '3scale/backend/storage_async/client'
 require '3scale/backend/storage_async/pipeline'
 require '3scale/backend/storage_async/resque_extensions'

lib/3scale/backend/storage_async/client.rb

@@ -1,3 +1,4 @@
+require 'concurrent'
 require 'async/io'
 require 'async/redis/client'
 
@@ -13,6 +14,7 @@ module StorageAsync
       # the Storage instance behaves likes the redis-rb client.
       class Client
         include Configurable
+        include Methods
 
         DEFAULT_HOST = 'localhost'.freeze
         private_constant :DEFAULT_HOST
@@ -46,124 +48,13 @@ def initialize(opts)
           port ||= DEFAULT_PORT
 
           endpoint = Async::IO::Endpoint.tcp(host, port)
-          @redis_async = Async::Redis::Client.new(
+          @redis_async = Concurrent::ThreadLocalVar.new{ Async::Redis::Client.new(
             endpoint, limit: opts[:max_connections]
-          )
-          @building_pipeline = false
+          )}
         end
 
-        # Now we are going to define the methods to run redis commands
-        # following the interface of the redis-rb lib.
-        #
-        # These are the different cases:
-        #   1) Methods that can be called directly. For example SET:
-        #      @redis_async.call('SET', some_key)
-        #   2) Methods that need to be "boolified". These are methods for which
-        #      redis-rb returns a boolean, but redis just returns an integer.
-        #      For example, Redis returns 0 or 1 for the EXISTS command, but
-        #      redis-rb transforms that into a boolean.
-        #   3) There are a few methods that need to be treated differently and
-        #      do not fit in any of the previous categories. For example, SSCAN
-        #      which accepts a hash of options in redis-rb.
-        #
-        # All of this might be simplified a bit in the future using the
-        # "methods" in async-redis
-        # https://github.com/socketry/async-redis/tree/master/lib/async/redis/methods
-        # but there are some commands missing, so for now, that's not an option.
-
-        METHODS_TO_BE_CALLED_DIRECTLY = [
-          :del,
-          :exists,
-          :expire,
-          :expireat,
-          :flushdb,
-          :get,
-          :hset,
-          :hmget,
-          :incr,
-          :incrby,
-          :keys,
-          :llen,
-          :lpop,
-          :lpush,
-          :lrange,
-          :ltrim,
-          :mget,
-          :ping,
-          :rpush,
-          :sadd,
-          :scard,
-          :setex,
-          :smembers,
-          :srem,
-          :sunion,
-          :ttl,
-          :zcard,
-          :zrangebyscore,
-          :zremrangebyscore,
-          :zrevrange
-        ].freeze
-        private_constant :METHODS_TO_BE_CALLED_DIRECTLY
-
-        METHODS_TO_BE_CALLED_DIRECTLY.each do |method|
-          define_method(method) do |*args|
-            @redis_async.call(method, *args.flatten)
-          end
-        end
-
-        METHODS_TO_BOOLIFY = [
-          :exists?,
-          :sismember,
-          :sadd?,
-          :srem?,
-          :zadd
-        ].freeze
-        private_constant :METHODS_TO_BOOLIFY
-
-        METHODS_TO_BOOLIFY.each do |method|
-          command = method.to_s.delete('?')
-          define_method(method) do |*args|
-            @redis_async.call(command, *args.flatten) > 0
-          end
-        end
-
-        def blpop(*args)
-          call_args = ['BLPOP'] + args
-
-          # redis-rb accepts a Hash as last arg that can contain :timeout.
-          if call_args.last.is_a? Hash
-            timeout = call_args.pop[:timeout]
-            call_args << timeout
-          end
-
-          @redis_async.call(*call_args.flatten)
-        end
-
-        def set(key, val, opts = {})
-          args = ['SET', key, val]
-
-          args += ['EX', opts[:ex]] if opts[:ex]
-          args << 'NX' if opts[:nx]
-
-          @redis_async.call(*args)
-        end
-
-        def sscan(key, cursor, opts = {})
-          args = ['SSCAN', key, cursor]
-
-          args += ['MATCH', opts[:match]] if opts[:match]
-          args += ['COUNT', opts[:count]] if opts[:count]
-
-          @redis_async.call(*args)
-        end
-
-        def scan(cursor, opts = {})
-          args = ['SCAN', cursor]
-
-          args += ['MATCH', opts[:match]] if opts[:match]
-          args += ['COUNT', opts[:count]] if opts[:count]
-
-          @redis_async.call(*args)
+        def call(*args)
+          @redis_async.value.call(*args)
         end
 
         # This method allows us to send pipelines like this:
@@ -178,31 +69,13 @@ def pipelined(&block)
           # There's an important limitation: this assumes that the fiber will
           # not yield in the block.
 
-          # When running a nested pipeline, we just need to continue
-          # accumulating commands.
-          if @building_pipeline
-            block.call self
-            return
-          end
-
-          @building_pipeline = true
-
-          original = @redis_async
           pipeline = Pipeline.new
-          @redis_async = pipeline
-
-          begin
-            block.call self
-          ensure
-            @redis_async = original
-            @building_pipeline = false
-          end
-
-          pipeline.run(original)
+          block.call pipeline
+          pipeline.run(@redis_async.value)
         end
 
         def close
-          @redis_async.close
+          @redis_async.value.close
         end
       end
 

lib/3scale/backend/storage_async/methods.rb

@@ -0,0 +1,123 @@
+# frozen_string_literal: true
+
+module ThreeScale
+  module Backend
+    module StorageAsync
+      module Methods
+        # Now we are going to define the methods to run redis commands
+        # following the interface of the redis-rb lib.
+        #
+        # These are the different cases:
+        #   1) Methods that can be called directly. For example SET:
+        #      @redis_async.call('SET', some_key)
+        #   2) Methods that need to be "boolified". These are methods for which
+        #      redis-rb returns a boolean, but redis just returns an integer.
+        #      For example, Redis returns 0 or 1 for the EXISTS command, but
+        #      redis-rb transforms that into a boolean.
+        #   3) There are a few methods that need to be treated differently and
+        #      do not fit in any of the previous categories. For example, SSCAN
+        #      which accepts a hash of options in redis-rb.
+        #
+        # All of this might be simplified a bit in the future using the
+        # "methods" in async-redis
+        # https://github.com/socketry/async-redis/tree/master/lib/async/redis/methods
+        # but there are some commands missing, so for now, that's not an option.
+
+        METHODS_TO_BE_CALLED_DIRECTLY = [
+          :del,
+          :exists,
+          :expire,
+          :expireat,
+          :flushdb,
+          :get,
+          :hset,
+          :hmget,
+          :incr,
+          :incrby,
+          :keys,
+          :llen,
+          :lpop,
+          :lpush,
+          :lrange,
+          :ltrim,
+          :mget,
+          :ping,
+          :rpush,
+          :sadd,
+          :scard,
+          :setex,
+          :smembers,
+          :srem,
+          :sunion,
+          :ttl,
+          :zcard,
+          :zrangebyscore,
+          :zremrangebyscore,
+          :zrevrange
+        ].freeze
+        private_constant :METHODS_TO_BE_CALLED_DIRECTLY
+
+        METHODS_TO_BE_CALLED_DIRECTLY.each do |method|
+          define_method(method) do |*args|
+            call(method, *args.flatten)
+          end
+        end
+
+        METHODS_TO_BOOLIFY = [
+          :exists?,
+          :sismember,
+          :sadd?,
+          :srem?,
+          :zadd
+        ].freeze
+        private_constant :METHODS_TO_BOOLIFY
+
+        METHODS_TO_BOOLIFY.each do |method|
+          command = method.to_s.delete('?')
+          define_method(method) do |*args|
+            call(command, *args.flatten) > 0
+          end
+        end
+
+        def blpop(*args)
+          call_args = ['BLPOP'] + args
+
+          # redis-rb accepts a Hash as last arg that can contain :timeout.
+          if call_args.last.is_a? Hash
+            timeout = call_args.pop[:timeout]
+            call_args << timeout
+          end
+
+          call(*call_args.flatten)
+        end
+
+        def set(key, val, opts = {})
+          args = ['SET', key, val]
+
+          args += ['EX', opts[:ex]] if opts[:ex]
+          args << 'NX' if opts[:nx]
+
+          call(*args)
+        end
+
+        def sscan(key, cursor, opts = {})
+          args = ['SSCAN', key, cursor]
+
+          args += ['MATCH', opts[:match]] if opts[:match]
+          args += ['COUNT', opts[:count]] if opts[:count]
+
+          call(*args)
+        end
+
+        def scan(cursor, opts = {})
+          args = ['SCAN', cursor]
+
+          args += ['MATCH', opts[:match]] if opts[:match]
+          args += ['COUNT', opts[:count]] if opts[:count]
+
+          call(*args)
+        end
+      end
+    end
+  end
+end

lib/3scale/backend/storage_async/pipeline.rb

@@ -5,6 +5,7 @@ module StorageAsync
       # This class accumulates commands and sends several of them in a single
       # request, instead of sending them one by one.
       class Pipeline
+        include Methods
 
         Error = Class.new StandardError
 
@@ -55,6 +56,12 @@ def call(*args)
           1
         end
 
+        def pipelined(&block)
+          # When running a nested pipeline, we just need to continue
+          # accumulating commands.
+          block.call self
+        end
+
         # Send to redis all the accumulated commands.
         # Returns an array with the result for each command in the same order
         # that they added with .call().