Class: ConsistentCluster::SyncClient

Inherits:
Object
  • Object
show all
Defined in:
lib/consistent-cluster/sync-client.rb

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ SyncClient 



12
13
14
15
16
17
18
19
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
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/consistent-cluster/sync-client.rb', line 12

def initialize(options)

  @syncingMutex = Mutex.new

  @data = {}
  @cluster = {}
  @node_register = {}

  replicas = options[:consistent_hashing_replicas] || 3
  @ring = ConsistentHashing::Ring.new([],replicas)

  @create_proc = options[:create_proc]
  @destroy_proc = options[:destroy_proc]
  @after_sync_proc = options[:after_sync_proc]
  @log_proc = options[:log_proc]

  @logger = if options[:log_file]
    Logger.new(options[:log_file])
  elsif STDERR || STDOUT
    Logger.new(STDERR || STDOUT)
  else
    Logger.new(options[:nil])
  end

  @last_sync_at = 0

  @path = options[:zookeeper_path]

  @zk = ZK.new(options[:zookeeper_service],reconnect: true)

  @zk.register(@path) do |event|
    sync_services
  end

  sync_services

  #on_state_change
  #on_connecting
  #on_expired_session
  @zk.on_connected do
    reconnect_callback
  end

  check_timer = options[:check_timer]
  check_timer ||= 60 #default 1 minutes gap

  start_check_timer_thread(check_timer)

  @shard_num = 0
end

Instance Method Details

#invoke_sync ⇒ Object 



89
90
91
 
# File 'lib/consistent-cluster/sync-client.rb', line 89

def invoke_sync
  sync_services
end

#last_sync_at ⇒ Object 



112
113
114
 
# File 'lib/consistent-cluster/sync-client.rb', line 112

def last_sync_at
  @last_sync_at
end

#local_version ⇒ Object 



93
94
95
 
# File 'lib/consistent-cluster/sync-client.rb', line 93

def local_version
  @data
end

#remote_version ⇒ Object 



97
98
99
100
101
102
103
104
105
106
 
# File 'lib/consistent-cluster/sync-client.rb', line 97

def remote_version
  app_names = get_app_names(watch: false)

  data = {}
  app_names.each do |app_name|
    app_content = get_app_content(app_name, watch: false)
    data[app_name] = app_content
  end
  data
end

#shard(key = nil) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/consistent-cluster/sync-client.rb', line 75

def shard(key=nil)
  cluster_sum = @cluster.length
  raise "no service available at #{@path}" if cluster_sum < 1
  if key
    point = @ring.point_for(key)
    server = @cluster[point.node]
  else
    @shard_num += 1
    @shard_num = @shard_num%cluster_sum
    server = @cluster.values[@shard_num]
  end
  server
end

#start_check_timer_thread(check_timer) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/consistent-cluster/sync-client.rb', line 63

def start_check_timer_thread(check_timer)
  return if check_timer <= 0
  Thread.new {
    while true
      sleep check_timer
      if !@syncingMutex.locked? && local_version != remote_version
        sync_services
      end
    end
  }
end

#zookeeper_path ⇒ Object 



108
109
110
 
# File 'lib/consistent-cluster/sync-client.rb', line 108

def zookeeper_path
  @path
end