Class: Prefab::ConfigClient

Inherits:
Object
  • Object
show all
Defined in:
lib/prefab/config_client.rb

Constant Summary collapse

RECONNECT_WAIT = 
5
DEFAULT_CHECKPOINT_FREQ_SEC = 
60
DEFAULT_S3CF_BUCKET = 
'http://d2j4ed6ti5snnd.cloudfront.net'

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(base_client, timeout) ⇒ ConfigClient

Returns a new instance of ConfigClient.



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/prefab/config_client.rb', line 7

def initialize(base_client, timeout)
  @base_client = base_client
  @base_client.log_internal Logger::DEBUG, "Initialize ConfigClient"
  @timeout = timeout
  @initialization_lock = Concurrent::ReadWriteLock.new

  @checkpoint_freq_secs = DEFAULT_CHECKPOINT_FREQ_SEC

  @config_loader = Prefab::ConfigLoader.new(@base_client)
  @config_resolver = Prefab::ConfigResolver.new(@base_client, @config_loader)

  @base_client.log_internal Logger::DEBUG, "Initialize ConfigClient: AcquireWriteLock"
  @initialization_lock.acquire_write_lock
  @base_client.log_internal Logger::DEBUG, "Initialize ConfigClient: AcquiredWriteLock"

  @cancellable_interceptor = Prefab::CancellableInterceptor.new(@base_client)

  @s3_cloud_front = ENV["PREFAB_S3CF_BUCKET"] || DEFAULT_S3CF_BUCKET

  load_checkpoint
  start_checkpointing_thread
end

Class Method Details

.value_to_delta(key, config_value, namespace = nil) ⇒ Object 



65
66
67
68
 
# File 'lib/prefab/config_client.rb', line 65

def self.value_to_delta(key, config_value, namespace = nil)
  Prefab::Config.new(key: [namespace, key].compact.join(":"),
                     rows: [Prefab::ConfigRow.new(value: config_value)])
end

Instance Method Details

#get(key) ⇒ Object 



36
37
38
39
40
 
# File 'lib/prefab/config_client.rb', line 36

def get(key)
  @initialization_lock.with_read_lock do
    @config_resolver.get(key)
  end
end

#get_config_obj(key) ⇒ Object 



70
71
72
73
74
 
# File 'lib/prefab/config_client.rb', line 70

def get_config_obj(key)
  @initialization_lock.with_read_lock do
    @config_resolver.get_config(key)
  end
end

#reset ⇒ Object 



56
57
58
59
 
# File 'lib/prefab/config_client.rb', line 56

def reset
  @base_client.reset!
  @_stub = nil
end

#start_streaming ⇒ Object 



30
31
32
33
34
 
# File 'lib/prefab/config_client.rb', line 30

def start_streaming
  @streaming = true
  # start_grpc_streaming_connection_thread(@config_loader.highwater_mark)
  start_sse_streaming_connection_thread(@config_loader.highwater_mark)
end

#to_s ⇒ Object 



61
62
63
 
# File 'lib/prefab/config_client.rb', line 61

def to_s
  @config_resolver.to_s
end

#upsert(key, config_value, namespace = nil, previous_key = nil) ⇒ Object 



42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/prefab/config_client.rb', line 42

def upsert(key, config_value, namespace = nil, previous_key = nil)
  raise "Key must not contain ':' set namespaces separately" if key.include? ":"
  raise "Namespace must not contain ':'" if namespace&.include?(":")
  config_delta = Prefab::ConfigClient.value_to_delta(key, config_value, namespace)
  upsert_req = Prefab::UpsertRequest.new(config_delta: config_delta)
  upsert_req.previous_key = previous_key if previous_key&.present?

  @base_client.request Prefab::ConfigService, :upsert, req_options: { timeout: @timeout }, params: upsert_req
  @base_client.stats.increment("prefab.config.upsert")
  @config_loader.set(config_delta)
  @config_loader.rm(previous_key) if previous_key&.present?
  @config_resolver.update
end