Class: EventMachine::Protocols::Couchbase::Client

Inherits:
Object
  • Object
show all
Includes:
EM::Deferrable
Defined in:
lib/em-couchbase/client.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Client

Returns a new instance of Client.



29
30
31
32
33
34
35
 
# File 'lib/em-couchbase/client.rb', line 29

def initialize
  @opaque = 0
  @nodes = []
  @admin_ports = []
  @packets = {}
  @upgrade_queue = EM::Queue.new
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



27
28
29
 
# File 'lib/em-couchbase/client.rb', line 27

def config
  @config
end

#nodes ⇒ Object (readonly)

Returns the value of attribute nodes.



26
27
28
 
# File 'lib/em-couchbase/client.rb', line 26

def nodes
  @nodes
end

Instance Method Details

#connect(options = {}) ⇒ Object

Parameters:

  • options (Hash) (defaults to: {})

Options Hash (options):

  • :hostname (String) — default: "localhost"
  • :port (Fixnum) — default: 8091
  • :pool (String) — default: "default"
  • :bucket (String) — default: "default"
  • :username (String) — default: nil
  • :password (String) — default: nil 


44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/em-couchbase/client.rb', line 44

def connect(options = {})
  @config_listener = ConfigurationListener.new
  @config_listener.on_error do |listener, error|
    if @admin_ports.empty?
      @on_error.call(self, error) if @on_error
    else
      options = options.merge(@admin_ports.shuffle!.pop)
      @config_listener.listen(options)
    end
  end
  @config_listener.on_upgrade do |config|
    @config = config
    config.nodes.each_with_index do |nn, ii|
      if nodes[ii] != nn
        nodes[ii].close_connection if nodes[ii]
        nodes[ii] = Node.connect(nn.merge(:client => self))
      end
    end
    @admin_ports = nodes.map do |node|
      host, port = node.admin.split(':')
      {:hostname => host, :port => port}
    end
    do_retry = lambda do |payload|
      opaque, packet = payload
      key, handler, raw = packet.values_at(:key, :handler, :raw)
      register_handler(opaque, key, handler)

      vbucket = raw[6..7].unpack("n").first
      if @config.vbucket_map_forward
        @config.vbucket_map[vbucket] = @config.vbucket_map_forward[vbucket].dup
      end
      node = @nodes[@config.vbucket_map[vbucket][0]]

      register_packet(opaque, raw)
      node.callback do
        node.send_data(raw)
      end
      @upgrade_queue.pop(&do_retry)
    end
    @upgrade_queue.pop(&do_retry)
    succeed
  end
  @config_listener.listen(options)
  self
end

#decr(key, options = {}, &block) ⇒ Object 



178
179
180
181
182
183
184
185
186
187
 
# File 'lib/em-couchbase/client.rb', line 178

def decr(key, options = {}, &block)
  callback do
    opaque = opaque_inc
    register_handler(opaque, key, block)
    vbucket, node = locate(key)
    node.callback do
      node.arithm(:decr, opaque, vbucket, key, options)
    end
  end
end

#get(*keys, &block) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/em-couchbase/client.rb', line 146

def get(*keys, &block)
  callback do
    if keys.last.is_a?(Hash)
      options = keys.last.pop
    end
    groups = keys.inject({}) do |acc, key|
      opaque = opaque_inc
      register_handler(opaque, key, block)
      vbucket, node = locate(key)
      acc[node] ||= []
      acc[node] << [opaque, vbucket, key]
      acc
    end
    groups.each do |node, tuple|
      node.callback do
        node.get(tuple, options)
      end
    end
  end
end

#incr(key, options = {}, &block) ⇒ Object 



167
168
169
170
171
172
173
174
175
176
 
# File 'lib/em-couchbase/client.rb', line 167

def incr(key, options = {}, &block)
  callback do
    opaque = opaque_inc
    register_handler(opaque, key, block)
    vbucket, node = locate(key)
    node.callback do
      node.arithm(:incr, opaque, vbucket, key, options)
    end
  end
end

#locate(key) ⇒ Fixnum

Locate node using vbucket distribution

Returns:

  • (Fixnum) —

    server index



128
129
130
131
132
133
 
# File 'lib/em-couchbase/client.rb', line 128

def locate(key)
  digest = Couchbase::Util.crc32_hash(key.to_s)
  mask = @config.vbucket_map.size - 1
  vbucket = digest & mask
  [vbucket, @nodes[@config.vbucket_map[vbucket][0]]]
end

#on_error(&callback) ⇒ Object 



90
91
92
 
# File 'lib/em-couchbase/client.rb', line 90

def on_error(&callback)
  @on_error = callback
end

#opaque_inc ⇒ Object (protected) 



197
198
199
200
 
# File 'lib/em-couchbase/client.rb', line 197

def opaque_inc
  @opaque += 1
  @opaque &= 0xffffffff # 32 bits
end

#register_handler(opaque, key, handler) ⇒ Object 



101
102
103
104
105
 
# File 'lib/em-couchbase/client.rb', line 101

def register_handler(opaque, key, handler)
  packet = (@packets[opaque] ||= {})
  packet[:key] = key
  packet[:handler] = handler
end

#register_packet(opaque, packet) ⇒ Object 



94
95
96
97
98
99
 
# File 'lib/em-couchbase/client.rb', line 94

def register_packet(opaque, packet)
  if packet.respond_to?(:force_encoding)
    packet.force_encoding(Encoding::BINARY)
  end
  (@packets[opaque] ||= {})[:raw] = packet
end

#retry(reason, opaque) ⇒ Object 



107
108
109
110
111
112
113
114
115
 
# File 'lib/em-couchbase/client.rb', line 107

def retry(reason, opaque)
  packet = @packets.delete(opaque)
  if packet
    case reason
    when :not_my_vbucket
      @upgrade_queue.push([opaque, packet])
    end
  end
end

#run_callback(opaque, result) ⇒ Object 



117
118
119
120
121
122
123
124
 
# File 'lib/em-couchbase/client.rb', line 117

def run_callback(opaque, result)
  packet = @packets.delete(opaque)
  key, handler = packet.values_at(:key, :handler)
  if handler.respond_to?(:call)
    result.key = key
    handler.call(result)
  end
end

#set(key, val, options = {}, &block) ⇒ Object 



135
136
137
138
139
140
141
142
143
144
 
# File 'lib/em-couchbase/client.rb', line 135

def set(key, val, options = {}, &block)
  callback do
    opaque = opaque_inc
    register_handler(opaque, key, block)
    vbucket, node = locate(key)
    node.callback do
      node.set(opaque, vbucket, key, val, options)
    end
  end
end

#unbind ⇒ Object (protected) 



191
192
193
194
195
 
# File 'lib/em-couchbase/client.rb', line 191

def unbind
  nodes.each do |node|
    nodes.disconnect
  end
end