/usr/lib/python2.7/dist-packages/zmq/devices/monitoredqueue.pxd is in python-zmq 14.0.1-1build2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 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 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | """MonitoredQueue class declarations.
Authors
-------
* MinRK
* Brian Granger
"""
#
# Copyright (c) 2010 Min Ragan-Kelley, Brian Granger
#
# This file is part of pyzmq, but is derived and adapted from zmq_queue.cpp
# originally from libzmq-2.1.6, used under LGPLv3
#
# pyzmq is free software; you can redistribute it and/or modify it under
# the terms of the Lesser GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# pyzmq is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# Lesser GNU General Public License for more details.
#
# You should have received a copy of the Lesser GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------
from libzmq cimport *
#-----------------------------------------------------------------------------
# MonitoredQueue C functions
#-----------------------------------------------------------------------------
cdef inline int _relay(void *insocket_, void *outsocket_, void *sidesocket_,
zmq_msg_t msg, zmq_msg_t side_msg, zmq_msg_t id_msg,
bint swap_ids) nogil:
cdef int rc
cdef int64_t flag_2
cdef int flag_3
cdef int flags
cdef bint more
cdef size_t flagsz
cdef void * flag_ptr
if ZMQ_VERSION_MAJOR < 3:
flagsz = sizeof (int64_t)
flag_ptr = &flag_2
else:
flagsz = sizeof (int)
flag_ptr = &flag_3
if swap_ids:# both router, must send second identity first
# recv two ids into msg, id_msg
rc = zmq_msg_recv(&msg, insocket_, 0)
if rc < 0: return rc
rc = zmq_msg_recv(&id_msg, insocket_, 0)
if rc < 0: return rc
# send second id (id_msg) first
#!!!! always send a copy before the original !!!!
rc = zmq_msg_copy(&side_msg, &id_msg)
if rc < 0: return rc
rc = zmq_msg_send(&side_msg, outsocket_, ZMQ_SNDMORE)
if rc < 0: return rc
rc = zmq_msg_send(&id_msg, sidesocket_, ZMQ_SNDMORE)
if rc < 0: return rc
# send first id (msg) second
rc = zmq_msg_copy(&side_msg, &msg)
if rc < 0: return rc
rc = zmq_msg_send(&side_msg, outsocket_, ZMQ_SNDMORE)
if rc < 0: return rc
rc = zmq_msg_send(&msg, sidesocket_, ZMQ_SNDMORE)
if rc < 0: return rc
while (True):
rc = zmq_msg_recv(&msg, insocket_, 0)
if rc < 0: return rc
# assert (rc == 0)
rc = zmq_getsockopt (insocket_, ZMQ_RCVMORE, flag_ptr, &flagsz)
if rc < 0: return rc
flags = 0
if ZMQ_VERSION_MAJOR < 3:
if flag_2:
flags |= ZMQ_SNDMORE
else:
if flag_3:
flags |= ZMQ_SNDMORE
# LABEL has been removed:
# rc = zmq_getsockopt (insocket_, ZMQ_RCVLABEL, flag_ptr, &flagsz)
# if flag_3:
# flags |= ZMQ_SNDLABEL
# assert (rc == 0)
rc = zmq_msg_copy(&side_msg, &msg)
if rc < 0: return rc
if flags:
rc = zmq_msg_send(&side_msg, outsocket_, flags)
if rc < 0: return rc
# only SNDMORE for side-socket
rc = zmq_msg_send(&msg, sidesocket_, ZMQ_SNDMORE)
if rc < 0: return rc
else:
rc = zmq_msg_send(&side_msg, outsocket_, 0)
if rc < 0: return rc
rc = zmq_msg_send(&msg, sidesocket_, 0)
if rc < 0: return rc
break
return rc
# the MonitoredQueue C function, adapted from zmq::queue.cpp :
cdef inline int c_monitored_queue (void *insocket_, void *outsocket_,
void *sidesocket_, zmq_msg_t *in_msg_ptr,
zmq_msg_t *out_msg_ptr, int swap_ids) nogil:
"""The actual C function for a monitored queue device.
See ``monitored_queue()`` for details.
"""
cdef zmq_msg_t msg
cdef int rc = zmq_msg_init (&msg)
cdef zmq_msg_t id_msg
rc = zmq_msg_init (&id_msg)
if rc < 0: return rc
cdef zmq_msg_t side_msg
rc = zmq_msg_init (&side_msg)
if rc < 0: return rc
cdef zmq_pollitem_t items [2]
items [0].socket = insocket_
items [0].fd = 0
items [0].events = ZMQ_POLLIN
items [0].revents = 0
items [1].socket = outsocket_
items [1].fd = 0
items [1].events = ZMQ_POLLIN
items [1].revents = 0
# I don't think sidesocket should be polled?
# items [2].socket = sidesocket_
# items [2].fd = 0
# items [2].events = ZMQ_POLLIN
# items [2].revents = 0
while (True):
# // Wait while there are either requests or replies to process.
rc = zmq_poll (&items [0], 2, -1)
if rc < 0: return rc
# // The algorithm below asumes ratio of request and replies processed
# // under full load to be 1:1. Although processing requests replies
# // first is tempting it is suspectible to DoS attacks (overloading
# // the system with unsolicited replies).
#
# // Process a request.
if (items [0].revents & ZMQ_POLLIN):
# send in_prefix to side socket
rc = zmq_msg_copy(&side_msg, in_msg_ptr)
if rc < 0: return rc
rc = zmq_msg_send(&side_msg, sidesocket_, ZMQ_SNDMORE)
if rc < 0: return rc
# relay the rest of the message
rc = _relay(insocket_, outsocket_, sidesocket_, msg, side_msg, id_msg, swap_ids)
if rc < 0: return rc
if (items [1].revents & ZMQ_POLLIN):
# send out_prefix to side socket
rc = zmq_msg_copy(&side_msg, out_msg_ptr)
if rc < 0: return rc
rc = zmq_msg_send(&side_msg, sidesocket_, ZMQ_SNDMORE)
if rc < 0: return rc
# relay the rest of the message
rc = _relay(outsocket_, insocket_, sidesocket_, msg, side_msg, id_msg, swap_ids)
if rc < 0: return rc
return rc
|