1
=== modified file 'doc/README'
2
--- doc/README	2009-08-15 10:27:56 +0000
3
+++ doc/README	2010-03-11 10:38:21 +0000
4
@@ -1,51 +1,60 @@
8
1
This is a quick post on getting started using the txAMQP library with RabbitMQ,
1
Here is some information you may need for getting started using the txAMQP
9
2
based on Dan Reverri's article "Getting started with RabbitMQ, Python, Twisted,
2
library with RabbitMQ, based on Dan Reverri's article "Getting started with
10
3
and txAMQP on Ubuntu 9.04" [http://www.apparatusproject.org/blog/?p=38]
3
RabbitMQ, Python, Twisted, and txAMQP on Ubuntu 9.04"
11
4
[http://www.apparatusproject.org/blog/?p=38]
12
4
5
13
5
Install RabbitMQ
6
Install RabbitMQ
14
7
----------------
15
6
	
8
	
16
7
If you're using Debian (or a derivative, like Ubuntu) and RabbitMQ is not
9
If you're using Debian (or a derivative, like Ubuntu) and RabbitMQ is not
17
8
included in your distribution, you can add RabbitMQ's APT repository:
10
included in your distribution, you can add RabbitMQ's APT repository:
18
9
11
19
10
http://www.rabbitmq.com/debian.html#apt
12
http://www.rabbitmq.com/debian.html#apt
20
11
13
22
12
Running RabbitMQ
14
Run RabbitMQ
23
15
------------
24
13
16
25
14
If you installed RabbitMQ using the Debian package, it will be automatically
17
If you installed RabbitMQ using the Debian package, it will be automatically
26
15
started. If not, you'll find an init.d file in /etc/init.d/rabbitmq-server
18
started. If not, you'll find an init.d file in /etc/init.d/rabbitmq-server
27
16
19
28
17
Install Twisted
20
Install Twisted
29
21
---------------
30
18
22
35
19
As of 2009-07-09, txAMQP is being developed using Twisted 8.2.0, it may
23
txAMQP was developed under Twisted 8.2.0, it may work with older versions,
36
20
work with older versions, but it's not guaranteed. In case you are using
24
but there is no guarantee.  txAMQP is known to run successfully on Twisted
37
21
Ubuntu, and the version available is not recent enough, you can install
25
9.0 and 10.0.  If you are using Ubuntu, and your version of Twisted is not
38
22
Twisted through its PPA:
26
recent enough, you can install Twisted from its PPA:
39
23
27
40
24
https://launchpad.net/~twisted-dev/+archive/ppa
28
https://launchpad.net/~twisted-dev/+archive/ppa
41
25
29
42
26
Install txAMQP
30
Install txAMQP
43
31
--------------
44
27
32
45
28
The instructions below install txAQMP from trunk. There are Ubuntu packages
33
The instructions below install txAQMP from trunk. There are Ubuntu packages
47
29
available if you prefer, refer to the following page for more details:
34
available if you prefer: refer to the following page for more details:
48
30
35
49
31
https://launchpad.net/~txamqpteam/+archive/ppa
36
https://launchpad.net/~txamqpteam/+archive/ppa
50
32
37
51
33
a. Install Bazaar
38
a. Install Bazaar
52
34
39
54
35
$ sudo apt-get install bzr
40
   $ sudo apt-get install bzr
55
36
41
56
37
b. Fetch txAQMP
42
b. Fetch txAQMP
57
38
43
66
39
$ cd ~
44
   $ bzr branch lp:txamqp txamqp
67
40
$ bzr branch lp:txamqp txamqp
45
68
41
46
   If you get the error "bzr: ERROR: Unknown repository format: 'Bazaar
69
42
If you get the error "bzr: ERROR: Unknown repository format: 'Bazaar
47
   RepositoryFormatKnitPack6 (bzr 1.9)" or similar, you'll need to upgrade
70
43
RepositoryFormatKnitPack6 (bzr 1.9)", you'll need to upgrade Bazaar. You may
48
   Bazaar. To do so, follow the instructions at: http://bazaar-vcs.org/Download
63
44
do so following the instructions at:
64
45
65
46
http://bazaar-vcs.org/Download
71
47
49
72
48
c. Install txAMQP
50
c. Install txAMQP
73
49
51
76
50
$ cd txamqp
52
   $ cd txamqp
77
51
$ sudo python setup.py install
53
   $ sudo python setup.py install
78
54
79
55
Run a txAMQP example
80
56
--------------------
81
57
82
58
You can confirm that txAMQP is properly installed and working correctly by
83
59
running the scripts in the src/examples/simple directory. See the README
84
60
file in that directory for detailed instructions.
85
52
61
86
=== modified file 'setup.py'
87
--- setup.py	2009-08-16 11:30:01 +0000
88
+++ setup.py	2010-03-11 10:38:21 +0000
89
@@ -10,7 +10,9 @@
90
10
    from setuptools import setup, find_packages
10
    from setuptools import setup, find_packages
91
11
except ImportError:
11
except ImportError:
92
12
    from distutils.core import setup
12
    from distutils.core import setup
94
13
    setupdict['packages'] = ['txamqp', 'txamqp.contrib', 'txamqp.contrib.thrift']
13
    setupdict['packages'] = ['txamqp',
95
14
                             'txamqp.contrib',
96
15
                             'txamqp.contrib.thrift']
97
14
    setupdict['package_dir'] = {
16
    setupdict['package_dir'] = {
98
15
        'txamqp': 'src/txamqp',
17
        'txamqp': 'src/txamqp',
99
16
        'txamqp.contrib': 'src/txamqp/contrib',
18
        'txamqp.contrib': 'src/txamqp/contrib',
100
17
19
101
=== modified file 'src/examples/simple/README'
102
--- src/examples/simple/README	2009-08-15 10:31:11 +0000
103
+++ src/examples/simple/README	2010-03-11 10:38:21 +0000
104
@@ -1,6 +1,10 @@
105
1
Instructions
1
Instructions
106
2
============
2
============
107
3
3
108
4
This directory contains a simple example of code that uses txAMQP to
109
5
connect to an AMQP broker, create an exchange, and then publish and receive
110
6
a number of messages.
111
7
112
4
Requirements
8
Requirements
113
5
------------
9
------------
114
6
10
115
@@ -14,7 +18,7 @@
116
14
   a) In one of them run the txconsumer.py script:
18
   a) In one of them run the txconsumer.py script:
117
15
      $ python txconsumer.py host port vhost username password path_to_spec
19
      $ python txconsumer.py host port vhost username password path_to_spec
118
16
      e.g. txconsumer.py localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml
20
      e.g. txconsumer.py localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml
120
17
   b) In the other one run the txpublisher.py script:
21
   b) In the other, run the txpublisher.py script:
121
18
      $ python txpublisher.py host port vhost username password path_to_spec content [count]
22
      $ python txpublisher.py host port vhost username password path_to_spec content [count]
122
19
      e.g. txpublisher.py localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml hello 1000
23
      e.g. txpublisher.py localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml hello 1000
123
20
24
124
21
25
125
=== modified file 'src/examples/simple/txconsumer.py'
126
--- src/examples/simple/txconsumer.py	2009-08-15 10:31:11 +0000
127
+++ src/examples/simple/txconsumer.py	2010-03-11 10:38:21 +0000
128
@@ -1,50 +1,64 @@
129
1
from twisted.internet.defer import inlineCallbacks
1
from twisted.internet.defer import inlineCallbacks
130
2
from twisted.internet import reactor
2
from twisted.internet import reactor
131
3
from twisted.internet.protocol import ClientCreator
3
from twisted.internet.protocol import ClientCreator
132
4
from twisted.python import log
133
4
from txamqp.protocol import AMQClient
5
from txamqp.protocol import AMQClient
135
5
from txamqp.client import TwistedDelegate
6
from txamqp.client import TwistedDelegate, Closed
136
6
import txamqp.spec
7
import txamqp.spec
137
7
8
138
8
@inlineCallbacks
9
@inlineCallbacks
139
9
def gotConnection(conn, username, password):
10
def gotConnection(conn, username, password):
140
10
    print "Connected to broker."
11
    print "Connected to broker."
142
11
    yield conn.authenticate(username, password)
12
143
13
    try:
144
14
        yield conn.authenticate(username, password)
145
15
    except Closed:
146
16
        print "Failed to authenticate with AMQP broker. Check that " \
147
17
              "the username and password you're supplying are correct."
148
18
        return
149
12
19
150
13
    print "Authenticated. Ready to receive messages"
20
    print "Authenticated. Ready to receive messages"
151
14
    chan = yield conn.channel(1)
21
    chan = yield conn.channel(1)
152
15
    yield chan.channel_open()
22
    yield chan.channel_open()
153
16
23
160
17
    yield chan.queue_declare(queue="chatrooms", durable=True, exclusive=False, auto_delete=False)
24
    # Create an exchange (the producer can do this too, with no ill effect).
161
18
    yield chan.exchange_declare(exchange="chatservice", type="direct", durable=True, auto_delete=False)
25
    yield chan.exchange_declare(exchange="chatservice", type="direct",
162
19
26
                                durable=True, auto_delete=False)
163
20
    yield chan.queue_bind(queue="chatrooms", exchange="chatservice", routing_key="txamqp_chatroom")
27
    
164
21
28
    yield chan.queue_declare(queue="chatrooms", durable=True,
165
22
    yield chan.basic_consume(queue='chatrooms', no_ack=True, consumer_tag="testtag")
29
                             exclusive=False, auto_delete=False)
166
30
167
31
    yield chan.queue_bind(queue="chatrooms", exchange="chatservice",
168
32
                          routing_key="txamqp_chatroom")
169
33
170
34
    yield chan.basic_consume(queue='chatrooms', no_ack=True,
171
35
                             consumer_tag="testtag")
172
23
36
173
24
    queue = yield conn.queue("testtag")
37
    queue = yield conn.queue("testtag")
174
25
38
175
26
    while True:
39
    while True:
176
27
        msg = yield queue.get()
40
        msg = yield queue.get()
178
28
        print 'Received: ' + msg.content.body + ' from channel #' + str(chan.id)
41
        print 'Received: %r from channel #%s' % (msg.content.body, chan.id)
179
29
        if msg.content.body == "STOP":
42
        if msg.content.body == "STOP":
180
30
            break
43
            break
181
31
44
182
45
    # Cancel consuming from the queue.
183
32
    yield chan.basic_cancel("testtag")
46
    yield chan.basic_cancel("testtag")
184
33
47
185
48
    # Close the channel.
186
34
    yield chan.channel_close()
49
    yield chan.channel_close()
187
35
50
188
51
    # Close the connection (must be done over channel 0).
189
36
    chan0 = yield conn.channel(0)
52
    chan0 = yield conn.channel(0)
190
37
191
38
    yield chan0.connection_close()
53
    yield chan0.connection_close()
192
39
54
193
40
    reactor.stop()
194
41
195
42
55
196
43
if __name__ == "__main__":
56
if __name__ == "__main__":
197
44
    import sys
57
    import sys
198
45
    if len(sys.argv) < 7:
58
    if len(sys.argv) < 7:
199
46
        print "%s host port vhost username password path_to_spec" % sys.argv[0]
59
        print "%s host port vhost username password path_to_spec" % sys.argv[0]
201
47
        print "e.g. %s localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml" % sys.argv[0]
60
        print "e.g. %s localhost 5672 / guest guest " \
202
61
              "../../specs/standard/amqp0-8.xml" % sys.argv[0]
203
48
        sys.exit(1)
62
        sys.exit(1)
204
49
63
205
50
    host = sys.argv[1]
64
    host = sys.argv[1]
206
@@ -58,8 +72,10 @@
207
58
    delegate = TwistedDelegate()
72
    delegate = TwistedDelegate()
208
59
73
209
60
    d = ClientCreator(reactor, AMQClient, delegate=delegate, vhost=vhost,
74
    d = ClientCreator(reactor, AMQClient, delegate=delegate, vhost=vhost,
211
61
        spec=spec).connectTCP(host, port)
75
                      spec=spec).connectTCP(host, port)
212
62
76
213
63
    d.addCallback(gotConnection, username, password)
77
    d.addCallback(gotConnection, username, password)
215
64
78
    d.addErrback(log.err)
216
79
    d.addBoth(lambda _: reactor.callLater(0, reactor.stop))
217
80
    
218
65
    reactor.run()
81
    reactor.run()
219
66
82
220
=== modified file 'src/examples/simple/txpublisher.py'
221
--- src/examples/simple/txpublisher.py	2009-08-15 10:31:11 +0000
222
+++ src/examples/simple/txpublisher.py	2010-03-11 10:38:21 +0000
223
@@ -1,51 +1,63 @@
224
1
from twisted.internet.defer import inlineCallbacks
1
from twisted.internet.defer import inlineCallbacks
225
2
from twisted.internet import reactor, task
2
from twisted.internet import reactor, task
226
3
from twisted.internet.protocol import ClientCreator
3
from twisted.internet.protocol import ClientCreator
227
4
from twisted.python import log
228
4
from txamqp.protocol import AMQClient
5
from txamqp.protocol import AMQClient
230
5
from txamqp.client import TwistedDelegate
6
from txamqp.client import TwistedDelegate, Closed
231
6
from txamqp.content import Content
7
from txamqp.content import Content
232
7
import txamqp.spec
8
import txamqp.spec
233
8
9
234
9
@inlineCallbacks
10
@inlineCallbacks
235
10
def gotConnection(conn, username, password, body, count=1):
11
def gotConnection(conn, username, password, body, count=1):
236
12
    """
237
13
238
14
    @param: conn is an instance of AMQClient, the protocol class that
239
15
    handles talking to AMQP.
240
16
    """
241
11
    print "Connected to broker."
17
    print "Connected to broker."
243
12
    yield conn.authenticate(username, password)
18
244
19
    try:
245
20
        yield conn.authenticate(username, password)
246
21
    except Closed:
247
22
        print "Failed to authenticate with AMQP broker. Check that " \
248
23
              "the username and password you're supplying are correct."
249
24
        return
250
13
25
251
14
    print "Authenticated. Ready to send messages"
26
    print "Authenticated. Ready to send messages"
252
15
    chan = yield conn.channel(1)
27
    chan = yield conn.channel(1)
253
16
    yield chan.channel_open()
28
    yield chan.channel_open()
254
17
29
274
18
    def send_messages():
30
    # Create an exchange (the consumer can do this too, with no ill effect).
275
19
        def message_iterator():
31
    yield chan.exchange_declare(exchange="chatservice", type="direct",
276
20
            for i in range(count):
32
                                durable=True, auto_delete=False)
277
21
                content = body + "-%d" % i
33
278
22
                msg = Content(content)
34
    def sender():
279
23
                msg["delivery mode"] = 2
35
        for text in ['%s-%d' % (body, i) for i in range(count)] + ['STOP']:
280
24
                chan.basic_publish(exchange="chatservice", content=msg, routing_key="txamqp_chatroom")
36
            msg = Content(text)
281
25
                print "Sending message: %s" % content
37
            msg["delivery mode"] = 2
282
26
                yield None
38
            print "Sending message: %r" % text
283
27
        return task.coiterate(message_iterator())
39
            yield chan.basic_publish(exchange="chatservice", content=msg,
284
28
40
                                     routing_key="txamqp_chatroom")
285
29
    yield send_messages()
41
            
286
30
    
42
    # Use a coiterator to send all the messages. This ensures the reactor
287
31
    stopToken = "STOP"
43
    # gets control once in a while to do the actual sending.
288
32
    msg = Content(stopToken)
44
    yield task.coiterate(sender())
289
33
    msg["delivery mode"] = 2
45
290
34
    chan.basic_publish(exchange="chatservice", content=msg, routing_key="txamqp_chatroom")
46
    # Close the channel.
272
35
    print "Sending message: %s" % stopToken
273
36
291
37
    yield chan.channel_close()
47
    yield chan.channel_close()
292
38
48
293
49
    # Close the connection (must be done over channel 0).
294
39
    chan0 = yield conn.channel(0)
50
    chan0 = yield conn.channel(0)
295
40
    yield chan0.connection_close()
51
    yield chan0.connection_close()
298
41
    
52
297
42
    reactor.stop()
299
43
    
53
    
300
44
if __name__ == "__main__":
54
if __name__ == "__main__":
301
45
    import sys
55
    import sys
302
46
    if len(sys.argv) < 8:
56
    if len(sys.argv) < 8:
305
47
        print "%s host port vhost username password path_to_spec content [count]" % sys.argv[0]
57
        print "%s host port vhost username password path_to_spec content " \
306
48
        print "e.g. %s localhost 5672 / guest guest ../../specs/standard/amqp0-8.xml hello 1000" % sys.argv[0]
58
              "[count]" % sys.argv[0]
307
59
        print "e.g. %s localhost 5672 / guest guest " \
308
60
              "../../specs/standard/amqp0-8.xml hello 1000" % sys.argv[0]
309
49
        sys.exit(1)
61
        sys.exit(1)
310
50
62
311
51
    host = sys.argv[1]
63
    host = sys.argv[1]
312
@@ -65,8 +77,10 @@
313
65
    delegate = TwistedDelegate()
77
    delegate = TwistedDelegate()
314
66
78
315
67
    d = ClientCreator(reactor, AMQClient, delegate=delegate, vhost=vhost,
79
    d = ClientCreator(reactor, AMQClient, delegate=delegate, vhost=vhost,
317
68
        spec=spec).connectTCP(host, port)
80
                      spec=spec).connectTCP(host, port)
318
69
81
319
70
    d.addCallback(gotConnection, username, password, content, count)
82
    d.addCallback(gotConnection, username, password, content, count)
320
83
    d.addErrback(log.err)
321
84
    d.addBoth(lambda _: reactor.callLater(0, reactor.stop))
322
71
85
323
72
    reactor.run()
86
    reactor.run()
324
73
87
325
=== modified file 'src/examples/thrift/README'
326
--- src/examples/thrift/README	2009-08-15 10:31:11 +0000
327
+++ src/examples/thrift/README	2010-03-11 10:38:21 +0000
328
@@ -1,6 +1,10 @@
329
1
Instructions
1
Instructions
330
2
============
2
============
331
3
3
332
4
This file contains some brief instructions on how to get and install Thrift
333
5
for Python, to use it to generate Twisted Python code to set-up a simple
334
6
service, and to run the example client and server code in this directory.
335
7
336
4
Dependencies
8
Dependencies
337
5
------------
9
------------
338
6
10
339
@@ -25,9 +29,9 @@
340
25
29
341
26
- General
30
- General
342
27
31
344
28
1. Download Thrift source code [1]
32
1. Download the Thrift source code from http://incubator.apache.org/thrift/
345
29
2. Check if it contains support for Twisted, you'll need revision 749795 or
33
2. Check if it contains support for Twisted, you'll need revision 749795 or
347
30
greater.
34
   greater.
348
31
3. Compile Thrift with suppport for Python.
35
3. Compile Thrift with suppport for Python.
349
32
4. Install Thrift.
36
4. Install Thrift.
350
33
37
351
@@ -37,27 +41,29 @@
352
37
1. Generate Thrift Python services:
41
1. Generate Thrift Python services:
353
38
    $ thrift --gen py:twisted shared.thrift
42
    $ thrift --gen py:twisted shared.thrift
354
39
    $ thrift --gen py:twisted tutorial.thrift
43
    $ thrift --gen py:twisted tutorial.thrift
355
44
356
40
2. Start a server:
45
2. Start a server:
357
41
    $ python server.py host port vhost username password path_to_spec
46
    $ python server.py host port vhost username password path_to_spec
358
47
359
42
    Note that you can start as many servers as you want at the same time.
48
    Note that you can start as many servers as you want at the same time.
361
43
    Incoming requests will be round-robined and the servers load will be
49
    Incoming requests will be round-robined; the servers' loads will be
362
44
    balanced automatically.
50
    balanced automatically.
363
51
364
45
3. Start a client:
52
3. Start a client:
365
46
    $ python client.py host port vhost username password path_to_spec
53
    $ python client.py host port vhost username password path_to_spec
366
47
54
383
48
Where:
55
   Where:
384
49
- host: The host where the AMQP broker is located.
56
    - host:         The host where the AMQP broker is located.
385
50
- port: The port where the AMQP broker listens on.
57
    - port:         The port the AMQP broker listens on.
386
51
- vhost: The virtual host (depending on your broker configuration, default
58
    - vhost:        The broker virtual host (depends on your broker
387
52
    in most cases is '/').
59
                    configuration, default in most cases is '/').
388
53
- username: The username to log in to the broker (default in most cases
60
    - username:     The username to log in to the broker with (default in
389
54
    is 'guest').
61
                    most cases is 'guest').
390
55
- password: The username to log in to the broker (default in most cases
62
    - password:     The password to log in to the broker with (default in
391
56
    is 'guest').
63
                    most cases is 'guest').
392
57
- path_to_spec: The path to the AMQP spec that you want to use. Keep in mind
64
    - path_to_spec: The path to the AMQP spec that you want to use. Note
393
58
    that depending on the broker you use, you will need a different spec:
65
                    that depending on the broker you use, you will need a
394
59
    - RabbitMQ 1.5.3: $TXAMQP_PATH/src/specs/standard/amqp0-8.xml
66
                    different spec:
395
60
    - OpenAMQ 1.3c5: $TXAMQP_PATH/src/specs/standard/amqp0-9.xml
67
                    RabbitMQ 1.5.3: $TXAMQP_PATH/src/specs/standard/amqp0-8.xml
396
61
    - Qpid M3 (Java): $TXAMQP_PATH/src/specs/qpid/amqp.0-8.xml
68
                    OpenAMQ 1.3c5: $TXAMQP_PATH/src/specs/standard/amqp0-9.xml
397
62
69
                    Qpid M3 (Java): $TXAMQP_PATH/src/specs/qpid/amqp.0-8.xml
382
63
1 - http://incubator.apache.org/thrift/
398
64
70
399
=== modified file 'src/txamqp/client.py'
400
--- src/txamqp/client.py	2009-08-08 17:54:09 +0000
401
+++ src/txamqp/client.py	2010-03-11 10:38:21 +0000
402
@@ -1,11 +1,18 @@
404
1
# coding: utf-8
1
# -*- coding: utf-8; -*-
405
2
406
2
from twisted.internet import defer
3
from twisted.internet import defer
407
3
from txamqp.delegate import Delegate
4
from txamqp.delegate import Delegate
408
4
5
409
6
410
5
class Closed(Exception):
7
class Closed(Exception):
411
6
    pass
8
    pass
412
7
9
413
10
414
8
class TwistedEvent(object):
11
class TwistedEvent(object):
415
12
    """A class that acts like an event that can be fired (via set). This is
416
13
    modeled on threading.Event, but in a Twisted style. Hence wait returns
417
14
    a Deferred instead of blocking.
418
15
    """
419
9
    def __init__(self):
16
    def __init__(self):
420
10
        self.deferred = defer.Deferred()
17
        self.deferred = defer.Deferred()
421
11
        self.alreadyCalled = False
18
        self.alreadyCalled = False
422
@@ -20,8 +27,19 @@
423
20
        deferred, self.deferred = self.deferred, defer.Deferred()
27
        deferred, self.deferred = self.deferred, defer.Deferred()
424
21
        deferred.callback(True)
28
        deferred.callback(True)
425
22
29
426
30
427
23
class TwistedDelegate(Delegate):
31
class TwistedDelegate(Delegate):
428
24
32
429
33
    """
430
34
    When the broker sends a method call to the client, such as
431
35
    basic_deliver, channel_close, channel_flow, etc., the method is routed
432
36
    to a TwistedDelegate.
433
37
434
38
    If you wanted to have something else receiving the messages from the
435
39
    broker, you would replace this class, which is what the Thrift contrib
436
40
    code does (see src/txamqp/contrib/thrift in the txAMQP distribution).
437
41
    """
438
42
439
25
    def connection_start(self, ch, msg):
43
    def connection_start(self, ch, msg):
440
26
        ch.connection_start_ok(mechanism=self.client.mechanism,
44
        ch.connection_start_ok(mechanism=self.client.mechanism,
441
27
                               response=self.client.response,
45
                               response=self.client.response,
442
28
46
443
=== modified file 'src/txamqp/codec.py'
444
--- src/txamqp/codec.py	2008-10-29 18:31:04 +0000
445
+++ src/txamqp/codec.py	2010-03-11 10:38:21 +0000
446
@@ -1,4 +1,6 @@
447
1
#!/usr/bin/env python
1
#!/usr/bin/env python
448
2
# -*- coding: utf-8; -*-
449
3
450
2
4
451
3
#
5
#
452
4
# Licensed to the Apache Software Foundation (ASF) under one
6
# Licensed to the Apache Software Foundation (ASF) under one
453
5
7
454
=== modified file 'src/txamqp/connection.py'
455
--- src/txamqp/connection.py	2009-04-27 17:00:08 +0000
456
+++ src/txamqp/connection.py	2010-03-11 10:38:21 +0000
457
@@ -1,3 +1,5 @@
458
1
# -*- coding: utf-8; -*-
459
2
460
1
#
3
#
461
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
462
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
463
4
6
464
=== modified file 'src/txamqp/content.py'
465
--- src/txamqp/content.py	2009-07-07 22:45:21 +0000
466
+++ src/txamqp/content.py	2010-03-11 10:38:21 +0000
467
@@ -1,3 +1,5 @@
468
1
# -*- coding: utf-8; -*-
469
2
470
1
#
3
#
471
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
472
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
473
4
6
474
=== modified file 'src/txamqp/delegate.py'
475
--- src/txamqp/delegate.py	2008-10-29 18:31:04 +0000
476
+++ src/txamqp/delegate.py	2010-03-11 10:38:21 +0000
477
@@ -1,3 +1,5 @@
478
1
# -*- coding: utf-8; -*-
479
2
480
1
#
3
#
481
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
482
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
483
@@ -20,7 +22,14 @@
484
20
import inspect
22
import inspect
485
21
from spec import pythonize
23
from spec import pythonize
486
22
24
488
23
class Delegate:
25
489
26
class Delegate(object):
490
27
  """
491
28
  This is the base class to which the txAMQP protocol class routes messages
492
29
  from the broker. It is subclassed by TwistedDelegate, or anything else
493
30
  that feels like it (for an example, see the ThriftTwistedDelegate class
494
31
  in src/txamqp/contrib/thrift/client.py).
495
32
  """
496
24
33
497
25
  def __init__(self):
34
  def __init__(self):
498
26
    self.handlers = {}
35
    self.handlers = {}
499
@@ -29,11 +38,23 @@
500
29
    self.invoke_all("init")
38
    self.invoke_all("init")
501
30
39
502
31
  def invoke_all(self, meth, *args, **kwargs):
40
  def invoke_all(self, meth, *args, **kwargs):
503
41
    """Call the init method (if any) in all superclasses of the subclass of
504
42
    Delegate. This currently doesn't call anything, but is here in case a
505
43
    future subclass needs an initialization mechanism.
506
44
    """
507
32
    for cls in inspect.getmro(self.__class__):
45
    for cls in inspect.getmro(self.__class__):
508
33
      if hasattr(cls, meth):
46
      if hasattr(cls, meth):
509
34
        getattr(cls, meth)(self, *args, **kwargs)
47
        getattr(cls, meth)(self, *args, **kwargs)
510
35
48
511
36
  def dispatch(self, channel, message):
49
  def dispatch(self, channel, message):
512
50
    """
513
51
    Dispatch an incoming AMQP method call to the client's handler. The
514
52
    method to call is either in self.handlers or is looked up on self. In
515
53
    the latter case, the name of the method is composed of the AMQP class
516
54
    ('basic', 'exchange', 'queue', etc) and the AMQP method ('return',
517
55
    'deliver', 'declare', etc). I.e., full method names look like
518
56
    'basic_declare' etc.
519
57
    """
520
37
    method = message.method
58
    method = message.method
521
38
59
522
39
    try:
60
    try:
523
40
61
524
=== modified file 'src/txamqp/message.py'
525
--- src/txamqp/message.py	2008-10-29 18:31:04 +0000
526
+++ src/txamqp/message.py	2010-03-11 10:38:21 +0000
527
@@ -1,3 +1,5 @@
528
1
# -*- coding: utf-8; -*-
529
2
530
1
#
3
#
531
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
532
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
533
4
6
534
=== modified file 'src/txamqp/protocol.py'
535
--- src/txamqp/protocol.py	2009-11-13 18:50:13 +0000
536
+++ src/txamqp/protocol.py	2010-03-11 10:38:21 +0000
537
@@ -1,4 +1,5 @@
539
1
# coding: utf-8
1
# -*- coding: utf-8; -*-
540
2
541
2
from twisted.python import log
3
from twisted.python import log
542
3
from twisted.internet import defer, protocol
4
from twisted.internet import defer, protocol
543
4
from twisted.internet.task import LoopingCall
5
from twisted.internet.task import LoopingCall
544
@@ -17,10 +18,12 @@
545
17
class GarbageException(Exception):
18
class GarbageException(Exception):
546
18
    pass
19
    pass
547
19
20
548
20
# An AMQP channel is a virtual connection that shares the
549
21
# same socket with others channels. One can have many channels
550
22
# per connection
551
23
class AMQChannel(object):
21
class AMQChannel(object):
552
22
    """
553
23
    An AMQP channel is a virtual connection that shares the same socket
554
24
    with others channels. One can have many channels per connection.
555
25
556
26
    """
557
24
27
558
25
    def __init__(self, id, outgoing):
28
    def __init__(self, id, outgoing):
559
26
        self.id = id
29
        self.id = id
560
@@ -45,21 +48,33 @@
561
45
        payload = frame.payload
48
        payload = frame.payload
562
46
        if isinstance(payload, Method):
49
        if isinstance(payload, Method):
563
47
            if payload.method.response:
50
            if payload.method.response:
564
51
                # If the method is a response (to an earlier method call on
565
52
                # the broker, such as queue_declare), put the frame into
566
53
                # the responses queue.
567
48
                self.queue = self.responses
54
                self.queue = self.responses
568
49
            else:
55
            else:
569
56
                # This is a incoming message from AMQP that requires
570
57
                # processing (e.g., connection_tune, channel_flow,
571
58
                # basic_deliver). We put the incoming queue into the work
572
59
                # queue of the AMQClient protocol so that as it receives
573
60
                # subsequent frames (in order), it can put them into our
574
61
                # incoming queue, which the worker (in AMQClient) consumes
575
62
                # and builds into a complete incoming message for dispatch.
576
63
                # This makes sure that incoming frames are properly
577
64
                # separated by channel number.
578
50
                self.queue = self.incoming
65
                self.queue = self.incoming
579
51
                work.put(self.incoming)
66
                work.put(self.incoming)
580
52
        self.queue.put(frame)
67
        self.queue.put(frame)
581
53
68
582
54
    @defer.inlineCallbacks
69
    @defer.inlineCallbacks
584
55
    def invoke(self, method, args, content = None):
70
    def invoke(self, method, args, content=None):
585
56
        if self.closed:
71
        if self.closed:
586
57
            raise Closed(self.reason)
72
            raise Closed(self.reason)
587
58
        frame = Frame(self.id, Method(method, *args))
73
        frame = Frame(self.id, Method(method, *args))
588
59
        self.outgoing.put(frame)
74
        self.outgoing.put(frame)
589
60
75
590
61
        if method.content:
76
        if method.content:
592
62
            if content == None:
77
            if content is None:
593
63
                content = Content()
78
                content = Content()
594
64
            self.writeContent(method.klass, content, self.outgoing)
79
            self.writeContent(method.klass, content, self.outgoing)
595
65
80
596
@@ -90,16 +105,28 @@
597
90
105
598
91
    def writeContent(self, klass, content, queue):
106
    def writeContent(self, klass, content, queue):
599
92
        size = content.size()
107
        size = content.size()
601
93
        header = Frame(self.id, Header(klass, content.weight(), size, **content.properties))
108
        header = Frame(self.id, Header(klass, content.weight(), size,
602
109
                                       **content.properties))
603
94
        queue.put(header)
110
        queue.put(header)
604
95
        for child in content.children:
111
        for child in content.children:
605
96
            self.writeContent(klass, child, queue)
112
            self.writeContent(klass, child, queue)
607
97
        # should split up if content.body exceeds max frame size
113
        # Should split this up if content.body exceeds max frame size.
608
98
        if size > 0:
114
        if size > 0:
609
99
            queue.put(Frame(self.id, Body(content.body)))
115
            queue.put(Frame(self.id, Body(content.body)))
610
100
116
611
117
612
101
class FrameReceiver(protocol.Protocol, basic._PauseableMixin):
118
class FrameReceiver(protocol.Protocol, basic._PauseableMixin):
613
102
119
614
120
    """
615
121
    A message from AMQP takes the form of one or more frames. The
616
122
    FrameReceiver class reads the binary data from the transport and calls
617
123
    frameReceived (which must be implemented by a subclass) for each
618
124
    complete frame.
619
125
620
126
    This class can also send frames to the AMQP broker.
621
127
    
622
128
    """
623
129
624
103
    frame_mode = False
130
    frame_mode = False
625
104
    MAX_LENGTH = 4096
131
    MAX_LENGTH = 4096
626
105
    HEADER_LENGTH = 1 + 2 + 4 + 1
132
    HEADER_LENGTH = 1 + 2 + 4 + 1
627
@@ -134,7 +161,8 @@
628
134
        payload = Frame.DECODERS[frameType].decode(self.spec, c)
161
        payload = Frame.DECODERS[frameType].decode(self.spec, c)
629
135
        end = c.decode_octet()
162
        end = c.decode_octet()
630
136
        if end != self.FRAME_END:
163
        if end != self.FRAME_END:
632
137
            raise GarbageException('frame error: expected %r, got %r' % (self.FRAME_END, end))
164
            raise GarbageException('frame error: expected %r, got %r' %
633
165
                                   (self.FRAME_END, end))
634
138
        frame = Frame(channel, payload)
166
        frame = Frame(channel, payload)
635
139
        return frame
167
        return frame
636
140
168
637
@@ -151,7 +179,8 @@
638
151
        while self.frame_mode and not self.paused:
179
        while self.frame_mode and not self.paused:
639
152
            sz = len(self.__buffer) - self.HEADER_LENGTH
180
            sz = len(self.__buffer) - self.HEADER_LENGTH
640
153
            if sz >= 0:
181
            if sz >= 0:
642
154
                length, = struct.unpack("!I", self.__buffer[3:7]) # size = 4 bytes
182
                # size = 4 bytes:
643
183
                length, = struct.unpack("!I", self.__buffer[3:7])
644
155
                if sz >= length:
184
                if sz >= length:
645
156
                    packet = self.__buffer[:self.HEADER_LENGTH + length]
185
                    packet = self.__buffer[:self.HEADER_LENGTH + length]
646
157
                    self.__buffer = self.__buffer[self.HEADER_LENGTH + length:]
186
                    self.__buffer = self.__buffer[self.HEADER_LENGTH + length:]
647
@@ -196,10 +225,29 @@
648
196
        content = body.payload.content
225
        content = body.payload.content
649
197
        buf.write(content)
226
        buf.write(content)
650
198
        read += len(content)
227
        read += len(content)
652
199
    defer.returnValue(Content(buf.getvalue(), children, header.properties.copy()))
228
    defer.returnValue(Content(buf.getvalue(), children,
653
229
                              header.properties.copy()))
654
200
230
655
201
class AMQClient(FrameReceiver):
231
class AMQClient(FrameReceiver):
656
202
232
657
233
    """
658
234
    This is the main protocol class for talking to an AMQP broker.
659
235
    
660
236
    All communication with the broker takes place in a virtual 'channel'
661
237
    (of which there may be many), so this class makes use of a (dynamically
662
238
    constructed) factory class for producing channel instances.
663
239
664
240
    The flow of control here is not easy to follow! Frames processed by the
665
241
    FrameReceiver class (which we inherit from) are passed successively to
666
242
    self.frameReceived which (in self.processFrame) passes them to the
667
243
    appropriate channel instance. When the frame payload is a method call
668
244
    (as opposed to a response), the channel puts its incoming queue onto
669
245
    our self.work queue, which results in our self.worker passing it (the
670
246
    channel's incoming queue) to self.dispatch, which reads the method
671
247
    content (if any), builds the complete incoming message and passes it to
672
248
    the delegate (see TwistedDelegate in client.py) to make the appropriate
673
249
    method call."""
674
250
    
675
203
    channelClass = AMQChannel
251
    channelClass = AMQChannel
676
204
252
677
205
    # Max unreceived heartbeat frames. The AMQP standard says it's 3.
253
    # Max unreceived heartbeat frames. The AMQP standard says it's 3.
678
@@ -214,23 +262,62 @@
679
214
262
680
215
        self.vhost = vhost
263
        self.vhost = vhost
681
216
264
682
265
        # Make a factory class that can generate new channel class
683
266
        # instances.  This is done using type as the class creator because
684
267
        # one of two the superclasses of the class we're making was made
685
268
        # dynamically from the AMQP spec.
686
217
        self.channelFactory = type("Channel%s" % self.spec.klass.__name__,
269
        self.channelFactory = type("Channel%s" % self.spec.klass.__name__,
687
218
                                    (self.channelClass, self.spec.klass), {})
270
                                    (self.channelClass, self.spec.klass), {})
688
271
689
272
        # A dictionary of channel instances. As dictionaries are not thread
690
273
        # safe, also make a lock for accessing the dictionary.
691
219
        self.channels = {}
274
        self.channels = {}
692
220
        self.channelLock = defer.DeferredLock()
275
        self.channelLock = defer.DeferredLock()
693
221
276
694
277
        # A queue for sending messages to AMQP.
695
222
        self.outgoing = defer.DeferredQueue()
278
        self.outgoing = defer.DeferredQueue()
696
279
697
280
        # A queue of incoming work in progress. Some AMQP method calls are
698
281
        # composed of several incoming messages (which in turn may be
699
282
        # composed of several frames). The objects that will be put into
700
283
        # the work queue (by a channel instance) are themselves queues.
701
284
        # These will be consumed from to assemble an incoming message.
702
223
        self.work = defer.DeferredQueue()
285
        self.work = defer.DeferredQueue()
703
224
286
704
287
        # The started event will fire once the connection is established
705
288
        # and the initial handshake has completed (prior to authentication).
706
225
        self.started = TwistedEvent()
289
        self.started = TwistedEvent()
707
226
290
708
291
        # A dictionary to hold basic_deliver messages coming from AMQP.
709
292
        # Each of these messages contains a tag that the queue consumer may
710
293
        # specify (or else the broker assigns one) when it sends a
711
294
        # basic_consume to the broker. The tag is used as a key into the
712
295
        # self.queues dictionary. The queuelock is used as a lock for the
713
296
        # dictionary when need to modify it, due to the non-thread-safeness
714
297
        # of dictionaries in Python.
715
298
        self.queues = {}
716
227
        self.queueLock = defer.DeferredLock()
299
        self.queueLock = defer.DeferredLock()
717
300
718
301
        # There are two cases in which the broker may send a basic_return
719
302
        # message.  For messages that are marked as mandatory and/or
720
303
        # immediate, but that cannot be delivered as such. For details, see
721
304
        # the AMQP spec. The basic_return_queue is used to hold these
722
305
        # incoming messages.
723
228
        self.basic_return_queue = TimeoutDeferredQueue()
306
        self.basic_return_queue = TimeoutDeferredQueue()
724
229
307
727
230
        self.queues = {}
308
        # Prepare to write outgoing messages to the broker by having our
728
231
309
        # writer method await the first outgoing message on the outgoing
729
310
        # queue. Thereafter it will arrange for itself to be called on
730
311
        # subsequent outgoing messages.
731
232
        self.outgoing.get().addCallback(self.writer)
312
        self.outgoing.get().addCallback(self.writer)
732
313
733
314
        # Prepare to deal with incoming messages from the broker by having
734
315
        # our worker method await the first set of incoming messages on the
735
316
        # work queue. Thereafter it will arrange for itself to be called on
736
317
        # subsequent incoming messages.
737
233
        self.work.get().addCallback(self.worker)
318
        self.work.get().addCallback(self.worker)
738
319
739
320
        # Arrange for heartbeat traffic, if requested.
740
234
        self.heartbeatInterval = heartbeat
321
        self.heartbeatInterval = heartbeat
741
235
        if self.heartbeatInterval > 0:
322
        if self.heartbeatInterval > 0:
742
236
            if clock is None:
323
            if clock is None:
743
@@ -296,10 +383,22 @@
744
296
        self.outgoing.get().addCallback(self.writer)
383
        self.outgoing.get().addCallback(self.writer)
745
297
384
746
298
    def worker(self, queue):
385
    def worker(self, queue):
747
386
        """
748
387
        @param queue: This is a queue from a channel instance created by
749
388
        self.channelFactory (i.e., normally the incoming queue of an
750
389
        AMQChannel subclass. See the dispatch() method of AMQChannel for
751
390
        the work.put call that puts a channel's queue into our self.work
752
391
        queue, which results in the channel's queue arriving here as the
753
392
        argument).
754
393
        """
755
299
        d = self.dispatch(queue)
394
        d = self.dispatch(queue)
756
395
757
396
        # Set things up so that we (i.e., self.worker) also handle future
758
397
        # work arriving on the self.work queue.
759
300
        def cb(ign):
398
        def cb(ign):
760
301
            self.work.get().addCallback(self.worker)
399
            self.work.get().addCallback(self.worker)
761
302
        d.addCallback(cb)
400
        d.addCallback(cb)
762
401
        
763
303
        d.addErrback(self.close)
402
        d.addErrback(self.close)
764
304
403
765
305
    @defer.inlineCallbacks
404
    @defer.inlineCallbacks
766
@@ -315,7 +414,7 @@
767
315
        message = Message(payload.method, payload.args, content)
414
        message = Message(payload.method, payload.args, content)
768
316
        self.delegate.dispatch(channel, message)
415
        self.delegate.dispatch(channel, message)
769
317
416
771
318
    # As soon as we connect to the target AMQP broker, send the init string
417
    # When we connect to the target AMQP broker, send the init string.
772
319
    def connectionMade(self):
418
    def connectionMade(self):
773
320
        self.sendInitString()
419
        self.sendInitString()
774
321
        self.setFrameMode()
420
        self.setFrameMode()
775
@@ -339,7 +438,8 @@
776
339
            self.reschedule_checkHB()
438
            self.reschedule_checkHB()
777
340
439
778
341
    @defer.inlineCallbacks
440
    @defer.inlineCallbacks
780
342
    def authenticate(self, username, password, mechanism='AMQPLAIN', locale='en_US'):
441
    def authenticate(self, username, password, mechanism='AMQPLAIN',
781
442
                     locale='en_US'):
782
343
        if self.check_0_8():
443
        if self.check_0_8():
783
344
            response = {"LOGIN": username, "PASSWORD": password}
444
            response = {"LOGIN": username, "PASSWORD": password}
784
345
        else:
445
        else:
785
@@ -374,4 +474,3 @@
786
374
            if self.checkHB.active():
474
            if self.checkHB.active():
787
375
                self.checkHB.cancel()
475
                self.checkHB.cancel()
788
376
        self.close(reason)
476
        self.close(reason)
789
377
790
378
477
791
=== modified file 'src/txamqp/queue.py'
792
--- src/txamqp/queue.py	2009-05-14 15:03:24 +0000
793
+++ src/txamqp/queue.py	2010-03-11 10:38:21 +0000
794
@@ -1,4 +1,5 @@
796
1
# coding: utf-8
1
# -*- coding: utf-8; -*-
797
2
798
2
from twisted.internet.defer import DeferredQueue
3
from twisted.internet.defer import DeferredQueue
799
3
4
800
4
class Empty(Exception):
5
class Empty(Exception):
801
5
6
802
=== modified file 'src/txamqp/spec.py'
803
--- src/txamqp/spec.py	2009-07-17 16:04:44 +0000
804
+++ src/txamqp/spec.py	2010-03-11 10:38:21 +0000
805
@@ -1,3 +1,5 @@
806
1
# -*- coding: utf-8; -*-
807
2
808
1
#
3
#
809
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
810
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
811
4
6
812
=== modified file 'src/txamqp/testlib.py'
813
--- src/txamqp/testlib.py	2009-08-08 18:03:05 +0000
814
+++ src/txamqp/testlib.py	2010-03-11 10:38:21 +0000
815
@@ -1,3 +1,5 @@
816
1
# -*- coding: utf-8; -*-
817
2
818
1
#
3
#
819
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
820
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
821
4
6
822
=== modified file 'src/txamqp/xmlutil.py'
823
--- src/txamqp/xmlutil.py	2009-07-09 21:28:09 +0000
824
+++ src/txamqp/xmlutil.py	2010-03-11 10:38:21 +0000
825
@@ -1,3 +1,5 @@
826
1
# -*- coding: utf-8; -*-
827
2
828
1
#
3
#
829
2
# Licensed to the Apache Software Foundation (ASF) under one
4
# Licensed to the Apache Software Foundation (ASF) under one
830
3
# or more contributor license agreements.  See the NOTICE file
5
# or more contributor license agreements.  See the NOTICE file
Status:	Approved
Approved by:	Thomas Herve on 2010-10-17
Approved revision:	43
Proposed branch:	lp:~terrycojones/txamqp/lots-of-comments
Merge into:	lp:txamqp
Diff against target:	830 lines (+305/-101) 17 files modified doc/README (+29/-20) setup.py (+3/-1) src/examples/simple/README (+5/-1) src/examples/simple/txconsumer.py (+31/-15) src/examples/simple/txpublisher.py (+40/-26) src/examples/thrift/README (+25/-19) src/txamqp/client.py (+19/-1) src/txamqp/codec.py (+2/-0) src/txamqp/connection.py (+2/-0) src/txamqp/content.py (+2/-0) src/txamqp/delegate.py (+22/-1) src/txamqp/message.py (+2/-0) src/txamqp/protocol.py (+115/-16) src/txamqp/queue.py (+2/-1) src/txamqp/spec.py (+2/-0) src/txamqp/testlib.py (+2/-0) src/txamqp/xmlutil.py (+2/-0)
To merge this branch:	bzr merge lp:~terrycojones/txamqp/lots-of-comments
Related bugs:	Link a bug report
Reviewer	Review Type	Date Requested	Status
Thomas Herve		2010-03-10	Approve on 2010-10-17
Review via email: mp+21048@code.launchpad.net