Class: LaunchDarkly::Impl::Integrations::Redis::RedisFeatureStoreCore

Inherits:
Object
  • Object
show all
Defined in:
lib/ldclient-rb/impl/integrations/redis_impl.rb

Overview

Internal implementation of the Redis feature store, intended to be used with CachingStoreWrapper.

Since:

  • 5.5.0

Instance Method Summary collapse

Constructor Details

#initialize(opts) ⇒ RedisFeatureStoreCore

Returns a new instance of RedisFeatureStoreCore.

Since:

  • 5.5.0



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 20

def initialize(opts)
  if !REDIS_ENABLED
    raise RuntimeError.new("can't use Redis feature store because one of these gems is missing: redis, connection_pool")
  end

  @redis_opts = opts[:redis_opts] || Hash.new
  if opts[:redis_url]
    @redis_opts[:url] = opts[:redis_url]
  end
  if !@redis_opts.include?(:url)
    @redis_opts[:url] = LaunchDarkly::Integrations::Redis::default_redis_url
  end
  max_connections = opts[:max_connections] || 16
  @pool = opts[:pool] || ConnectionPool.new(size: max_connections) do
    ::Redis.new(@redis_opts)
  end
  # shutdown pool on close unless the client passed a custom pool and specified not to shutdown
  @pool_shutdown_on_close = (!opts[:pool] || opts.fetch(:pool_shutdown_on_close, true))
  @prefix = opts[:prefix] || LaunchDarkly::Integrations::Redis::default_prefix
  @logger = opts[:logger] || Config.default_logger
  @test_hook = opts[:test_hook]  # used for unit tests, deliberately undocumented

  @stopped = Concurrent::AtomicBoolean.new(false)

  with_connection do |redis|
    @logger.info("RedisFeatureStore: using Redis instance at #{redis.connection[:host]}:#{redis.connection[:port]} \
      and prefix: #{@prefix}")
  end
end

Instance Method Details

#get_all_internal(kind) ⇒ Object

Since:

  • 5.5.0



73
74
75
76
77
78
79
80
81
82
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 73

def get_all_internal(kind)
  fs = {}
  with_connection do |redis|
    hashfs = redis.hgetall(items_key(kind))
    hashfs.each do |k, json_item|
      fs[k.to_sym] = Model.deserialize(kind, json_item)
    end
  end
  fs
end

#get_internal(kind, key) ⇒ Object

Since:

  • 5.5.0



67
68
69
70
71
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 67

def get_internal(kind, key)
  with_connection do |redis|
    get_redis(redis, kind, key)
  end
end

#init_internal(all_data) ⇒ Object

Since:

  • 5.5.0



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 50

def init_internal(all_data)
  count = 0
  with_connection do |redis|
    redis.multi do |multi|
      all_data.each do |kind, items|
        multi.del(items_key(kind))
        count = count + items.count
        items.each do |key, item|
          multi.hset(items_key(kind), key, Model.serialize(kind,item))
        end
      end
      multi.set(inited_key, inited_key)
    end
  end
  @logger.info { "RedisFeatureStore: initialized with #{count} items" }
end

#initialized_internal? ⇒ Boolean

Returns:

  • (Boolean)

Since:

  • 5.5.0



116
117
118
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 116

def initialized_internal?
  with_connection { |redis| redis.exists?(inited_key) }
end

#stop ⇒ Object

Since:

  • 5.5.0



120
121
122
123
124
125
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 120

def stop
  if @stopped.make_true
    return unless @pool_shutdown_on_close
    @pool.shutdown { |redis| redis.close }
  end
end

#upsert_internal(kind, new_item) ⇒ Object

Since:

  • 5.5.0



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/ldclient-rb/impl/integrations/redis_impl.rb', line 84

def upsert_internal(kind, new_item)
  base_key = items_key(kind)
  key = new_item[:key]
  try_again = true
  final_item = new_item
  while try_again
    try_again = false
    with_connection do |redis|
      redis.watch(base_key) do
        old_item = get_redis(redis, kind, key)
        before_update_transaction(base_key, key)
        if old_item.nil? || old_item[:version] < new_item[:version]
          result = redis.multi do |multi|
            multi.hset(base_key, key, Model.serialize(kind, new_item))
          end
          if result.nil?
            @logger.debug { "RedisFeatureStore: concurrent modification detected, retrying" }
            try_again = true
          end
        else
          final_item = old_item
          action = new_item[:deleted] ? "delete" : "update"
          @logger.warn { "RedisFeatureStore: attempted to #{action} #{key} version: #{old_item[:version]} \
    in '#{kind[:namespace]}' with a version that is the same or older: #{new_item[:version]}" }
        end
        redis.unwatch
      end
    end
  end
  final_item
end