Class: ZK::ZKEventMachine::Client

Inherits:
Client::Base
  • Object
show all
Includes:
Deferred::Accessors, Logging, Unixisms
Defined in:
lib/z_k/z_k_event_machine/client.rb

Constant Summary collapse

DEFAULT_TIMEOUT = 
10

Instance Method Summary collapse

Methods included from Unixisms

#mkdir_p, #rm_rf

Constructor Details

#initialize(host, opts = {}) ⇒ Client

Takes same options as ZK::Client::Base



56
57
58
59
60
61
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 56

def initialize(host, opts={})
  @host = host
  @event_handler  = EventHandlerEM.new(self)
  @closing        = false
  register_default_event_handlers!
end

Instance Method Details

#children(path, opts = {}, &block) ⇒ Object 



150
151
152
153
154
155
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 150

def children(path, opts={}, &block)
  Callback.new_children_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, opts.merge(:callback => cb))
  end
end

#close!(&blk) ⇒ Object Also known as: close 



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 79

def close!(&blk)
  on_close(&blk)
  return on_close if @closing
  @closing = true

  if @cnx
    logger.debug { "#{self.class.name}: in close! clearing event_handler" }
    event_handler.clear!

    logger.debug { "#{self.class.name}: calling @cnx.close" }
    @cnx.close do
      logger.debug { "firing on_close handler" }
      on_close.succeed
      @cnx = nil
    end
  else
    on_close.succeed
  end

  on_close
end

#connect(&blk) ⇒ Object

open a ZK connection, attach it to the reactor. returns an EM::Deferrable that will be called when the connection is ready for use



66
67
68
69
70
71
72
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 66

def connect(&blk)
  # XXX: maybe move this into initialize, need to figure out how to schedule it properly
  @cnx ||= (
    ZookeeperEM::Client.new(@host, DEFAULT_TIMEOUT, event_handler.get_default_watcher_block)
  )
  @cnx.on_attached(&blk)
end

#create(path, data = '', opts = {}, &block) ⇒ Object 



114
115
116
117
118
119
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 114

def create(path, data='', opts={}, &block)
  Callback.new_create_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, data, opts.merge(:callback => cb))
  end
end

#delete(path, opts = {}, &block) ⇒ Object 



143
144
145
146
147
148
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 143

def delete(path, opts={}, &block)
  Callback.new_delete_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, opts.merge(:callback => cb))
  end
end

#exists?(path, opts = {}, &block) ⇒ Boolean

Returns:

  • (Boolean) 


139
140
141
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 139

def exists?(path, opts={}, &block)
  stat(path, opts.merge(:cb_style => 'exists'), &block)
end

#get(path, opts = {}, &block) ⇒ Callback

get data at path, optionally enabling a watch on the node

Returns:

  • (Callback) —

    returns a Callback which is an EM::Deferred (so you can assign callbacks/errbacks) see Callback::Base for discussion



107
108
109
110
111
112
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 107

def get(path, opts={}, &block)
  Callback.new_get_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, opts.merge(:callback => cb))
  end
end

#get_acl(path, opts = {}, &block) ⇒ Object 



157
158
159
160
161
162
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 157

def get_acl(path, opts={}, &block)
  Callback.new_get_acl_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, opts.merge(:callback => cb))
  end
end

#on_close ⇒ Deferred::Default

called back once the connection has been closed.

Returns:

  • (Deferred::Default) 


53
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 53

deferred_event :close

#on_connected ⇒ Deferred::Default

Note:

this is experimental currently. This may or may not fire for the initial connection.

Registers a one-shot callback for the ZOO_CONNECTED_STATE event.

it’s purpose is to warn an already-existing client with watches that a connection has been re-established (with session information saved). From the ZooKeeper Programmers’ Guide:

If you are using watches, you must look for the connected watch event.
When a ZooKeeper client disconnects from a server, you will not receive
notification of changes until reconnected. If you are watching for a
znode to come into existance, you will miss the event if the znode is
created and deleted while you are disconnected.

once this deferred has been fired, it will be replaced with a new deferred, so callbacks must be re-registered, and should be re-registered within the callback to avoid missing events

Returns:

  • (Deferred::Default) 


41
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 41

deferred_event :connected

#on_connection_lost ⇒ Deferred::Default

If we get a ZK::Exceptions::ConnectionLoss exeption back from any call, or a EXPIRED_SESSION_STATE event, we will call back any handlers registered here with the exception instance as the argument.

once this deferred has been fired, it will be replaced with a new deferred, so callbacks must be re-registered, and should be re-registered within the callback to avoid missing events

Returns:

  • (Deferred::Default) 


20
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 20

deferred_event :connection_lost

#session_id ⇒ Fixnum

Returns The underlying connection’s session_id.

Returns:

  • (Fixnum) —

    The underlying connection’s session_id



172
173
174
175
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 172

def session_id
  return nil unless @cnx
  @cnx.session_id
end

#session_passwd ⇒ String

Returns The underlying connection’s session passwd (an opaque value).

Returns:

  • (String) —

    The underlying connection’s session passwd (an opaque value)



178
179
180
181
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 178

def session_passwd
  return nil unless @cnx
  @cnx.session_passwd
end

#set(path, data, opts = {}, &block) ⇒ Object 



121
122
123
124
125
126
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 121

def set(path, data, opts={}, &block)
  Callback.new_set_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, data, opts.merge(:callback => cb))
  end
end

#set_acl(path, acls, opts = {}, &block) ⇒ Object 



164
165
166
167
168
169
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 164

def set_acl(path, acls, opts={}, &block)
  Callback.new_set_acl_cb(block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, acls, opts.merge(:callback => cb))
  end
end

#stat(path, opts = {}, &block) ⇒ Object 



128
129
130
131
132
133
134
135
136
137
 
# File 'lib/z_k/z_k_event_machine/client.rb', line 128

def stat(path, opts={}, &block)
  cb_style = opts.delete(:cb_style) { |_| 'stat' }

  meth = :"new_#{cb_style}_cb"

  Callback.__send__(meth, block) do |cb|
    cb.errback(&method(:connection_lost_hook))
    super(path, opts.merge(:callback => cb))
  end
end