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
178
179
180
181
182
183
|
"""
kombu.transport.zookeeper
=========================
Zookeeper transport.
:copyright: (c) 2010 - 2012 by Mahendra M.
:license: BSD, see LICENSE for more details.
**Synopsis**
- Connects to a zookeeper node as <server>:<port>/<vhost>
The <vhost> becomes the base for all the other znodes. So we can use
it like a vhost
- A queue is a znode under the <vhost> path
- Creates a new sequential node under the queue and writes the message to it
- If priority is used, we will use it in the node name, so that higher
priority messages are picked up first
- Keep consuming messages from the top of the queue, till we
are able to delete a particular message. If deletion raises a
NoNode exception, we try again with the next message
**References**
- https://zookeeper.apache.org/doc/trunk/recipes.html#sc_recipes_Queues
- http://bit.ly/cZHf9g
**Limitations**
- A queue cannot handle more than 2^32 messages. This is an internal
limitation with zookeeper. This has to be handled internally in this
module.
"""
from __future__ import absolute_import
import socket
from anyjson import loads, dumps
from kombu.exceptions import StdConnectionError, StdChannelError
from kombu.five import Empty
from . import virtual
try:
import kazoo
KZ_CONNECTION_ERRORS = (
kazoo.zkclient.SystemErrorException,
kazoo.zkclient.ConnectionLossException,
kazoo.zkclient.MarshallingErrorException,
kazoo.zkclient.UnimplementedException,
kazoo.zkclient.OperationTimeoutException,
kazoo.zkclient.NoAuthException,
kazoo.zkclient.InvalidACLException,
kazoo.zkclient.AuthFailedException,
kazoo.zkclient.SessionExpiredException,
)
KZ_CHANNEL_ERRORS = (
kazoo.zkclient.RuntimeInconsistencyException,
kazoo.zkclient.DataInconsistencyException,
kazoo.zkclient.BadArgumentsException,
kazoo.zkclient.MarshallingErrorException,
kazoo.zkclient.UnimplementedException,
kazoo.zkclient.OperationTimeoutException,
kazoo.zkclient.ApiErrorException,
kazoo.zkclient.NoNodeException,
kazoo.zkclient.NoAuthException,
kazoo.zkclient.NodeExistsException,
kazoo.zkclient.NoChildrenForEphemeralsException,
kazoo.zkclient.NotEmptyException,
kazoo.zkclient.SessionExpiredException,
kazoo.zkclient.InvalidCallbackException,
)
except ImportError:
kazoo = None # noqa
KZ_CONNECTION_ERRORS = KZ_CHANNEL_ERRORS = () # noqa
DEFAULT_PORT = 2181
__author__ = 'Mahendra M <mahendra.m@gmail.com>'
class Channel(virtual.Channel):
_client = None
def _get_queue(self, queue):
return '/%s' % queue
def _put(self, queue, message, **kwargs):
try:
priority = message['properties']['delivery_info']['priority']
except KeyError:
priority = 0
msg_id = '%s/msg-%02d' % (self._get_queue(queue), priority % 10)
self.client.create(msg_id, dumps(message), sequence=True)
def _get_msg(self, queue, msgs):
msgs.sort() # this is a bad hack, but required
for msg_id in msgs:
msg_id = '%s/%s' % (queue, msg_id)
try:
message, headers = self.client.get(msg_id)
self.client.delete(msg_id)
except kazoo.zkclient.NoNodeException:
pass # Someone has got this message
else:
return loads(message)
raise Empty()
def _get(self, queue):
queue = self._get_queue(queue)
msgs = self.client.get_children(queue)
return self._get_msg(queue, msgs)
def _purge(self, queue):
failures = 0
queue = self._get_queue(queue)
for count, msg_id in enumerate(self.client.get_children(queue)):
try:
self.client.delete('%s/%s' % (queue, msg_id))
except kazoo.zkclient.NoNodeException:
failures += 1
return count - failures
def _delete(self, queue, *args, **kwargs):
if self._has_queue(queue):
queue = self._get_queue(queue)
self._purge(queue)
self.client.delete(queue)
def _size(self, queue):
_, meta = self.client.get(self._get_queue(queue))
return meta['numChildren']
def _new_queue(self, queue, **kwargs):
if not self._has_queue(queue):
self.client.create(self._get_queue(queue), '')
def _has_queue(self, queue):
return self.client.exists(self._get_queue(queue)) is not None
def _open(self):
conninfo = self.connection.client
port = conninfo.port or DEFAULT_PORT
conn_str = '%s:%s' % (conninfo.hostname, port)
conn_str += '/' + conninfo.virtual_host[0:-1]
conn = kazoo.ZooKeeperClient(conn_str)
conn.connect(timeout=conninfo.connect_timeout)
return conn
@property
def client(self):
if self._client is None:
self._client = self._open()
return self._client
class Transport(virtual.Transport):
Channel = Channel
polling_interval = 1
default_port = DEFAULT_PORT
connection_errors = (StdConnectionError, ) + KZ_CONNECTION_ERRORS
channel_errors = (StdChannelError, socket.error) + KZ_CHANNEL_ERRORS
driver_type = 'zookeeper'
driver_name = 'kazoo'
def __init__(self, *args, **kwargs):
if kazoo is None:
raise ImportError('The kazoo library is not installed')
super(Transport, self).__init__(*args, **kwargs)
def driver_version(self):
return kazoo.__version__
|
