Class: ShareDataWatcher::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/share-data-watcher/connection.rb

Overview

Etcd connection client

Constant Summary collapse

DEFAULT_FETCH_TIMEOUT = 
1

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(endpoints) ⇒ Connection

Returns a new instance of Connection.



12
13
14
 
# File 'lib/share-data-watcher/connection.rb', line 12

def initialize(endpoints)
  @conn = Etcdv3.new(endpoints: endpoints, allow_reconnect: false)
end

Instance Attribute Details

#conn ⇒ Object (readonly)

Returns the value of attribute conn.



8
9
10
 
# File 'lib/share-data-watcher/connection.rb', line 8

def conn
  @conn
end

Instance Method Details

#connecting_endpoint ⇒ Object 



16
17
18
 
# File 'lib/share-data-watcher/connection.rb', line 16

def connecting_endpoint
  conn.conn.connection.endpoint.to_s
end

#fetch(key, prefix: false, timeout: nil) ⇒ Object 



40
41
42
43
44
45
46
47
48
 
# File 'lib/share-data-watcher/connection.rb', line 40

def fetch(key, prefix: false, timeout: nil)
  opt = {}
  opt[:range_end] = increase_one_bit(key) if prefix
  opt[:timeout] = timeout || DEFAULT_FETCH_TIMEOUT
  res = conn.get(key, opt)
  revision = res.header.revision
  kvs = res.kvs
  block_given? ? yield(revision, kvs) : [revision, kvs]
end

#healthcheck ⇒ Object 



20
21
22
23
24
25
 
# File 'lib/share-data-watcher/connection.rb', line 20

def healthcheck
  conn.version
  true
rescue StandardError
  false
end

#watch(key, prefix: true, start_revision: nil) ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/share-data-watcher/connection.rb', line 27

def watch(key, prefix: true, start_revision: nil)
  opt = {
    timeout: Etcdv3::Watch::INFINITE_FUTURE
  }
  opt[:range_end] = increase_one_bit(key) if prefix
  opt[:start_revision] = start_revision if start_revision
  conn.watch(key, opt) do |events|
    events.each do |event|
      yield(event.type, event.kv)
    end
  end
end