Class: CZTop::Monitor

Inherits:
Object
  • Object
show all
Defined in:
lib/cztop/monitor.rb

Overview

Note:

The monitor must be created before the events you want to observe, and closed before closing the monitored socket.

Monitors ZMQ socket events (connections, disconnections, etc.) via CZMQ's zmonitor actor.

Defined Under Namespace

Classes: Event

Constant Summary collapse

EVENTS =

All supported ZMQ socket monitoring events.

%w[
  CONNECTED
  CONNECT_DELAYED
  CONNECT_RETRIED
  LISTENING
  BIND_FAILED
  ACCEPTED
  ACCEPT_FAILED
  CLOSED
  CLOSE_FAILED
  DISCONNECTED
  MONITOR_STOPPED
].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket, *events, verbose: false) ⇒ Monitor

Creates a new monitor for the given socket.

Parameters:

  • socket (CZTop::Socket) —

    the socket to monitor

  • events (Array<String>) —

    event names to listen for (default: all)

  • verbose (Boolean) (defaults to: false) —

    enable CZMQ verbose logging



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/cztop/monitor.rb', line 40

def initialize(socket, *events, verbose: false)
  zmonitor_fn = CZMQ::FFI::ZMONITOR_FN
  raise 'zmonitor not available in this CZMQ build' unless zmonitor_fn

  @actor_ptr = CZMQ::FFI.zactor_new(zmonitor_fn, socket.to_ptr)
  HasFFIDelegate.raise_zmq_err if @actor_ptr.null?

  @closed = false

  prevent_leak_ptr = ::FFI::MemoryPointer.new(:pointer)
  prevent_leak_ptr.write_pointer(@actor_ptr)
  ObjectSpace.define_finalizer(self, self.class._make_destructor(prevent_leak_ptr))

  CZMQ::FFI.zstr_send(@actor_ptr, 'VERBOSE') if verbose

  events = EVENTS if events.empty?
  events.each do |ev|
    CZMQ::FFI.zstr_sendm(@actor_ptr, 'LISTEN')
    CZMQ::FFI.zstr_send(@actor_ptr, ev)
  end

  CZMQ::FFI.zstr_send(@actor_ptr, 'START')
  CZMQ::FFI.zsock_wait(@actor_ptr)
end

Class Method Details

._make_destructor(prevent_leak_ptr) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



122
123
124
 
# File 'lib/cztop/monitor.rb', line 122

def self._make_destructor(prevent_leak_ptr)
  ->(_id) { CZMQ::FFI.zactor_destroy(prevent_leak_ptr) }
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Closes the monitor. Must be called before closing the monitored socket.



102
103
104
105
106
107
108
109
110
 
# File 'lib/cztop/monitor.rb', line 102

def close
  return if @closed

  pp = ::FFI::MemoryPointer.new(:pointer)
  pp.write_pointer(@actor_ptr)
  CZMQ::FFI.zactor_destroy(pp)
  @closed = true
  ObjectSpace.undefine_finalizer(self)
end

#closed? ⇒ Boolean

Returns whether the monitor has been closed.

Returns:

  • (Boolean) —

    whether the monitor has been closed



115
116
117
 
# File 'lib/cztop/monitor.rb', line 115

def closed?
  @closed
end

#receive(timeout: 0.1) ⇒ Event?

Receives the next monitoring event.

Parameters:

  • timeout (Numeric) (defaults to: 0.1) —

    timeout in seconds

Returns:

  • (Event, nil) —

    the event, or nil on timeout



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/cztop/monitor.rb', line 71

def receive(timeout: 0.1)
  CZMQ::FFI.zsock_set_rcvtimeo(@actor_ptr, (timeout * 1000).to_i)

  zmsg_ptr = CZMQ::FFI.zmsg_recv(@actor_ptr)
  return nil if zmsg_ptr.null?

  frames = []
  frame_ptr = CZMQ::FFI.zmsg_first(zmsg_ptr)
  until frame_ptr.null?
    data = CZMQ::FFI.zframe_data(frame_ptr)
    size = CZMQ::FFI.zframe_size(frame_ptr)
    frames << data.read_bytes(size).force_encoding(Encoding::UTF_8)
    frame_ptr = CZMQ::FFI.zmsg_next(zmsg_ptr)
  end

  pp = ::FFI::MemoryPointer.new(:pointer)
  pp.write_pointer(zmsg_ptr)
  CZMQ::FFI.zmsg_destroy(pp)

  Event.new(
    name:         frames[0],
    endpoint:     frames[1] || '',
    peer_address: frames[2]
  )
end