tnc / php-amqplib Goto Github PK

When I run a consumer and process some messages the next fails and goes to the queue again so it fails again when I start the consumer

server, version: 8.0, properties: capabilities=(), copyright=Copyright (C) 2007-2011 VMware, Inc., information=Licensed under the MPL. See http://www.rabbitmq.com/, platform=Erlang/OTP, product=RabbitMQ, version=2.5.1

0000 01 00 00 00 00 00 78 00 0A 00 0B 00 00 00 38 07 ......x. ......8.
0010 6C 69 62 72 61 72 79 53 00 00 00 13 50 48 50 20 libraryS ....PHP
0020 53 69 6D 70 6C 65 20 41 4D 51 50 20 6C 69 62 0F Simple A MQP lib.
0030 6C 69 62 72 61 72 79 5F 76 65 72 73 69 6F 6E 53 library_ versionS
0040 00 00 00 03 30 2E 31 08 41 4D 51 50 4C 41 49 4E ....0.1. AMQPLAIN
0050 00 00 00 25 05 4C 4F 47 49 4E 53 00 00 00 06 76 ...%.LOG INS....v
0060 6F 63 61 6C 69 08 50 41 53 53 57 4F 52 44 53 00 ocali.PA SSWORDS.
0070 00 00 06 76 6F 63 61 6C 69 05 65 6E 5F 55 53 CE ...vocal i.en_US▒

< 10,11: Connection.start_ok
waiting for 10,20, 10,30
waiting for a new frame

10,30: Connection.tune
< [hex]:
0000 01 00 00 00 00 00 0C 00 0A 00 1F FF FF 00 02 00 ........ ...▒▒...
0010 00 00 00 CE ...▒

< 10,31: Connection.tune_ok
< [hex]:
0000 01 00 00 00 00 00 08 00 0A 00 28 01 2F 00 00 CE ........ ..(./..▒

< 10,40: Connection.open
waiting for 10,41, 10,50
waiting for a new frame

10,41: Connection.open_ok
Open OK! known_hosts:
using channel_id: 1
< [hex]:
0000 01 00 01 00 00 00 05 00 14 00 0A 00 CE ........ ....▒

< 20,10: Channel.open
waiting for 20,11
waiting for a new frame

20,11: Channel.open_ok
Channel open
< [hex]:
0000 01 00 01 00 00 00 23 00 28 00 0A 00 00 10 70 72 ......#. (.....pr
0010 6F 63 65 73 73 2D 77 61 76 2D 66 69 6C 65 06 64 ocess-wa v-file.d
0020 69 72 65 63 74 02 00 00 00 00 CE irect... ..▒

< 40,10: Channel.exchange_declare
waiting for 40,11
waiting for a new frame

40,11: Channel.exchange_declare_ok
< [hex]:
0000 01 00 01 00 00 00 1C 00 32 00 0A 00 00 10 70 72 ........ 2.....pr
0010 6F 63 65 73 73 2D 77 61 76 2D 66 69 6C 65 02 00 ocess-wa v-file..
0020 00 00 00 CE ...▒

< 50,10: Channel.queue_declare
waiting for 50,11
waiting for a new frame

50,11: Channel.queue_declare_ok
< [hex]:
0000 01 00 01 00 00 00 2E 00 32 00 14 00 00 10 70 72 ........ 2.....pr
0010 6F 63 65 73 73 2D 77 61 76 2D 66 69 6C 65 10 70 ocess-wa v-file.p
0020 72 6F 63 65 73 73 2D 77 61 76 2D 66 69 6C 65 00 rocess-w av-file.
0030 00 00 00 00 00 CE .....▒

< 50,20: Channel.queue_bind
waiting for 50,21
waiting for a new frame

50,21: Channel.queue_bind_ok
< [hex]:
0000 01 00 01 00 00 00 3E 00 3C 00 14 00 00 10 70 72 ......>. <.....pr
0010 6F 63 65 73 73 2D 77 61 76 2D 66 69 6C 65 25 50 ocess-wa v-file%P
0020 48 50 50 52 4F 43 45 53 53 5F 6F 74 69 6C 69 6F HPPROCES S_otilio
0030 2E 6E 65 6F 73 69 73 74 65 63 2E 64 6F 6D 5F 31 .neosist ec.dom_1
0040 38 38 32 39 00 CE 8829.▒

< 60,20: Channel.basic_consume
waiting for 60,21
waiting for a new frame

60,21: Channel.basic_consume_ok
waiting for any method
waiting for a new frame
60,60: Channel.basic_deliver
waiting for a new frame
waiting for a new frame
New msg: {"id":65}
using channel_id: 2
< [hex]:
0000 01 00 02 00 00 00 05 00 14 00 0A 00 CE ........ ....▒

< 20,10: Channel.open
waiting for 20,11
waiting for a new frame

20,11: Channel.open_ok
Channel open
< [hex]:
0000 01 00 02 00 00 00 2B 00 28 00 0A 00 00 18 70 72 ......+. (.....pr
0010 6F 63 65 73 73 2D 74 72 61 6E 73 6C 61 74 69 6F ocess-tr anslatio
0020 6E 2D 74 65 78 74 06 64 69 72 65 63 74 02 00 00 n-text.d irect...
0030 00 00 CE ..▒

< 40,10: Channel.exchange_declare
waiting for 40,11
waiting for a new frame

40,11: Channel.exchange_declare_ok
< [hex]:
0000 01 00 02 00 00 00 21 00 3C 00 28 00 00 18 70 72 ......!. <.(...pr
0010 6F 63 65 73 73 2D 74 72 61 6E 73 6C 61 74 69 6F ocess-tr anslatio
0020 6E 2D 74 65 78 74 00 00 CE n-text.. ▒

< 60,40: Channel.basic_publish
< [hex]:
0000 02 00 02 00 00 00 1A 00 3C 00 00 00 00 00 00 00 ........ <.......
0010 00 00 09 90 00 0A 74 65 78 74 2F 70 6C 61 69 6E ...▒..te xt/plain
0020 02 CE .▒

< [hex]:
0000 03 00 02 00 00 00 09 7B 22 69 64 22 3A 36 35 7D .......{ "id":65}
0010 CE ▒

< [hex]:
0000 01 00 01 00 00 00 0D 00 3C 00 50 00 00 00 00 00 ........ <.P.....
0010 00 00 01 00 CE ....▒

< 60,80: Channel.basic_ack
waiting for any method
waiting for a new frame

[Exception]
Expecting AMQP method, received frame type: 3

I'm using the library inside a code that runs on hiphop. Hiphop, in order to optimize, doesn't call the __destruct() method of an object when removing the objects at the end of the script. This results in AMQPConnection to get destroyed without its __destruct() method is called.

I wanted to know if is it possible to move the code that exists in the __destruct() method to the close method of AMQPConnection class? That would make the library compatible with hiphop. But it would render the connection object unreusable once the close would be performed, that would be bad if the client wants to pool the connection objects on close. Any suggestions?

Some times I need get more than one items in queue to work.

diff --git a/amqp.inc b/amqp.inc
the actual code only checks for false as the return value of fwrite in AMQPConnection::write($data). sometimes, fwrite() does not return false but 0 bytes, and the while(true) makes php hang forever. this is the main concern here. the main loop and that fwrite is not returning false when an errno=32 happens. my solution (others may exist.. ):

index 1bf03b3..3607afe 100644
--- a/amqp.inc
+++ b/amqp.inc
@@ -436,7 +436,7 @@ class AMQPConnection extends AbstractChannel
$len = strlen($data);
while(true)
{

•        if(false === ($written = fwrite($this->sock, $data)))
  

•        if(false === ($written = fwrite($this->sock, $data)) || $written === 0)
       {
           throw new Exception ("Error sending data");
       }
  

thanks!

Hi

I have an issue regarding the consumption of "old" messages.

1. I'm producing many messages in my test
2. After stopping the producing script, I'm running 2 consumer scripts, with the same configuration using the basic_consume method.
3. I'm just seeing one consumer consuming messages on the queue. The other consumer idles after the basic_consume() call.
4. If I'm running the producer again, with 2 consumers, the first consumer works the old messages and the second consumer works the new messages.

I'm using one topic type Exchange ("testexchange") with one Queue bound ("testqueue") and routing ("routing.test").

What am I doing wrong?
I'd like to have all consumers work all messages in the queue.

Best
Daniel

$objAMQPChannel->exchange_declare('testexchange', 'topic', FALSE, TRUE, FALSE);
$objAMQPChannel->queue_declare('testqueue', false, true, false, false);
$objAMQPChannel->queue_bind('testqueue', 'testexchange', 'routing.test');

$objAMQPChannel->basic_consume('testqueue', posix_getpid(), FALSE, FALSE, FALSE, FALSE, $callbackFunctionName);
while(count($objAMQPChannel->callbacks)) {
$objAMQPChannel-->wait();
}

https://github.com/tnc/php-amqplib appears to have a bug described at http://code.google.com/p/php-amqplib/issues/detail?id=10

Attempted to use php-amqplib with qpid-java-0.8, and got an error message about php trying to allocate over 1GB of ram:

rrehbein@copper:~/dev/php-amqplib/demo$ php amqp_consumer.php

Fatal error: Allowed memory size of 134217728 bytes exhausted (tried to allocate 1342243073 bytes) in /home/rrehbein/dev/php-amqplib/amqp_wire.inc on line 293

Call Stack:

0.0003 633424 1. {main}() /home/rrehbein/dev/php-amqplib/demo/amqp_consumer.php:0
0.0038 1251136 2. AMQPConnection->__construct() /home/rrehbein/dev/php-amqplib/demo/amqp_consumer.php:21
0.0044 1255768 3. AbstractChannel->wait() /home/rrehbein/dev/php-amqplib/amqp.inc:384
0.0045 1255768 4. AbstractChannel->next_frame() /home/rrehbein/dev/php-amqplib/amqp.inc:280
0.0045 1255768 5. AMQPConnection->wait_channel() /home/rrehbein/dev/php-amqplib/amqp.inc:176
0.0045 1255768 6. AMQPConnection->wait_frame() /home/rrehbein/dev/php-amqplib/amqp.inc:566
0.0066 1264376 7. AMQPReader->read() /home/rrehbein/dev/php-amqplib/amqp.inc:549
0.0066 1264376 8. AMQPReader->rawread() /home/rrehbein/dev/php-amqplib/amqp_wire.inc:283
0.0066 1264600 9. fread() /home/rrehbein/dev/php-amqplib/amqp_wire.inc:293

System:
Debian Lenny, 64-bit PHP 5.3

Currently any basic_deliver messages received after a call to basic_cancel but before receiving basic_cancel_ok are queued, and then discarded without ack till the connection is closed.

Couple possible solutions:

• basic_cancel could reject basic_deliver messages for the cancelled consumer
• basic_cancel could continue processing basic_deliver messages for the canceled consumer till basic_cancel_ok is received. Could be done simply by adding "60,60" to the wait list in basic_cancel
• basic_cancel could have a mid-cancel callback option
• AMQPChannel could have an option for a fall-back callback option.

Running php-amqplib on a new system might yield the following error:

PHP Fatal error: Call to undefined function bcmod() in .../demo/amqp_wire.inc on line 74