Class: ZMQ::PollInterruptible

Inherits:
Poll
  • Object
show all
Defined in:
lib/0mq/poll_interruptible.rb

Overview

An interruptible version of Poll.

Instance Attribute Summary

Attributes inherited from Poll

#timeout

Instance Method Summary collapse

Methods inherited from Poll

poll, poll_nonblock, #run_nonblock

Constructor Details

#initialize(*sockets) ⇒ PollInterruptible

Creates the additional interruption objects and calls super Note that either #kill or #close MUST be called when done with the object. There is no automatic finalizer for this object.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/0mq/poll_interruptible.rb', line 10

def initialize(*sockets)
  @int_sock_rep = ZMQ::Socket.new ZMQ::REP
  @int_sock_req = ZMQ::Socket.new ZMQ::REQ
  
  # Choose an endpoint name that we can expect to be unique
  # so that they can be shared within the DefaultContext
  int_endpoint = "inproc://__PollInterruptible_int_"+hash.to_s(26)
  @int_sock_rep.bind    int_endpoint
  @int_sock_req.connect int_endpoint
  
  # Interruption blocks are stored here by key until #run receives them.
  # After each is run, the return value is stored here in its place.
  @interruptions = {}
  
  @dead = false
  
  super @int_sock_rep, *sockets
end

Instance Method Details

#close ⇒ Object

Permanently kill the Poll object This should be run once, when the Poll object is no longer needed. This should only be accessed when there is no poll thread running. Use #kill instead when there is a poll loop thread running.



90
91
92
93
94
95
96
97
 
# File 'lib/0mq/poll_interruptible.rb', line 90

def close
  return nil if @dead
  
  @int_sock_rep.close
  @int_sock_req.close
  
  @dead = true
end

#dead? ⇒ Boolean

Return true if the object has been killed or closed and cannot be run

Returns:

  • (Boolean) 


100
101
102
 
# File 'lib/0mq/poll_interruptible.rb', line 100

def dead?
  @dead
end

#interrupt(&block) ⇒ Object

Interrupt the running poll loop, but do not clean up. This should be run anytime to let the poller re-evaluate state, etc.. This should only be accessed from a thread other than the poll thread,

and only if the poll thread is running

If a block is given, it will be executed in the poll thread just

prior to the execution of the user block passed to {#run}.



59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/0mq/poll_interruptible.rb', line 59

def interrupt(&block)
  block ||= Proc.new { true }
  key = block.object_id.to_s 36
  
  @interruptions[key] = block # Store the block to be called
  
  @int_sock_req.send_string key # Signal an interruption to #run
  @int_sock_req.recv_array      # Wait until it has been handled by #run
  
  @interruptions.delete key # Return the stored result of the block
end

#kill ⇒ Object

Interrupt the running poll loop and permanently kill the Poll object This should be run once, when the Poll object is no longer needed. This should only be accessed from a thread other than the poll thread,

and only if the poll thread is running

Use #cleanup instead when there is no poll loop thread running.



76
77
78
79
80
81
82
83
84
 
# File 'lib/0mq/poll_interruptible.rb', line 76

def kill
  return nil if @dead
  
  @int_sock_req.send_array ["KILL"]
  @int_sock_req.recv_array
  @int_sock_req.close
  
  @dead = true
end

#run(&block) ⇒ Object

Same as Poll#run, but will yield [nil, nil] to the block if interrupted. Return value may be an empty hash if the poller was killed.



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/0mq/poll_interruptible.rb', line 31

def run(&block)
  raise "#{self} cannot run; it was permanently killed." if @dead
  
  super do |socket, revents|
    if socket == @int_sock_rep
      key, * = socket.recv_array
      kill = key == "KILL"
      
      # Call the user block of #interrupt and store the return value
      @interruptions[key] = @interruptions[key].call unless kill
      
      # Call the user block of #run
      block.call nil, nil if block
      
      socket.send_array ["OKAY"]
      @int_sock_rep.close if kill
    else
      block.call socket, revents if block
    end
  end.tap { |hash| hash.delete @int_sock_rep }
end