Class: ZK::EventHandlerSubscription::Actor

Inherits:
Base show all
Defined in:
lib/zk/event_handler_subscription/actor.rb

Overview

Stealing some ideas from Celluloid, this event handler subscription (basically, the wrapper around the user block), will spin up its own thread for delivery, and use a queue. This gives us the basis for better concurrency (event handlers run in parallel), but preserves the underlying behavior that a single-event-thread ZK gives us, which is that a single callback block is inherently serial. Without this, you have to make sure your callbacks are either synchronized, or totally reentrant, so that multiple threads could be calling your block safely (which is really difficult, and annoying).

Using this delivery mechanism means that the block still must not block forever, however each event will "wait its turn" and all callbacks will receive their events in the same order (which is what ZooKeeper guarantees), just perhaps at different times.

Constant Summary

Constants inherited from Base

Base::ALL_EVENTS, Base::ALL_EVENT_SET

Instance Attribute Summary

Attributes inherited from Base

#path

Attributes inherited from Subscription::Base

#callable, #parent

Instance Method Summary collapse

Methods included from Logger

#logger, wrapped_logger, wrapped_logger=

Methods inherited from Subscription::Base

#unregistered?, #unsubscribe

Constructor Details

#initialize(parent, path, callback, opts = {}) ⇒ Actor

Returns a new instance of Actor.



22
23
24
25
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 22

def initialize(parent, path, callback, opts={})
  super
  @threaded_callback = ThreadedCallback.new(@callable)
end

Instance Method Details

#async? ⇒ Boolean

Returns:

  • (Boolean) 


27
28
29
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 27

def async?
  true
end

#call(*args) ⇒ Object 



31
32
33
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 31

def call(*args)
  @threaded_callback.call(*args)
end

#close ⇒ Object

calls unsubscribe and shuts down



36
37
38
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 36

def close
  unregister
end

#pause_before_fork_in_parent ⇒ Object 



51
52
53
54
55
56
57
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 51

def pause_before_fork_in_parent
  synchronize do
    logger.debug { "#{self.class}##{__method__}" }
    @threaded_callback.pause_before_fork_in_parent
    super
  end
end

#reopen_after_fork! ⇒ Object 



45
46
47
48
49
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 45

def reopen_after_fork!
  logger.debug { "#{self.class}##{__method__}" }
  super
  @threaded_callback.reopen_after_fork!
end

#resume_after_fork_in_parent ⇒ Object 



59
60
61
62
63
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 59

def resume_after_fork_in_parent
  super
  logger.debug { "#{self.class}##{__method__}" }
  @threaded_callback.resume_after_fork_in_parent
end

#unregister ⇒ Object 



40
41
42
43
 
# File 'lib/zk/event_handler_subscription/actor.rb', line 40

def unregister
  super
  @threaded_callback.shutdown
end