php 连接rabbitmq例子含PhpAmqpLib库

<?php namespace PhpAmqpLib\Channel; use PhpAmqpLib\Channel\AbstractChannel; use PhpAmqpLib\Exception\AMQPProtocolChannelException; use PhpAmqpLib\Helper\MiscHelper; class AMQPChannel extends AbstractChannel { public $callbacks = array(); /** * * @var callable these parameters will be passed to function * in case of basic_return: * param int $reply_code * param string $reply_text * param string $exchange * param string $routing_key * param AMQPMessage $msg */ protected $basic_return_callback = null; public function __construct($connection, $channel_id=null, $auto_decode=true) { if ($channel_id == null) { $channel_id = $connection->get_free_channel_id(); } parent::__construct($connection, $channel_id); if ($this->debug) { MiscHelper::debug_msg("using channel_id: " . $channel_id); } $this->default_ticket = 0; $this->is_open = false; $this->active = true; // Flow control $this->alerts = array(); $this->callbacks = array(); $this->auto_decode = $auto_decode; $this->x_open(); } public function __destruct() { //TODO:???if($this->connection) // $this->close("destroying channel"); } /** * Tear down this object, after we've agreed to close with the server. */ protected function do_close() { $this->is_open = false; unset($this->connection->channels[$this->channel_id]); $this->channel_id = $this->connection = null; } /** * Only for AMQP0.8.0 * This method allows the server to send a non-fatal warning to * the client. This is used for methods that are normally * asynchronous and thus do not have confirmations, and for which * the server may detect errors that need to be reported. Fatal * errors are handled as channel or connection exceptions; non- * fatal errors are sent through this method. */ protected function channel_alert($args) { $reply_code = $args->read_short(); $reply_text = $args->read_shortstr(); $details = $args->read_table(); array_push($this->alerts,array($reply_code, $reply_text, $details)); } /** * request a channel close */ public function close($reply_code=0, $reply_text="", $method_sig=array(0, 0)) { list($class_id, $method_id, $args) = $this->protocolWriter->channelClose( $reply_code, $reply_text, $method_sig[0], $method_sig[1] ); $this->send_method_frame(array($class_id, $method_id), $args); return $this->wait(array( $this->waitHelper->get_wait('channel.close_ok') )); } protected function channel_close($args) { $reply_code = $args->read_short(); $reply_text = $args->read_shortstr(); $class_id = $args->read_short(); $method_id = $args->read_short(); $this->send_method_frame(array(20, 41)); $this->do_close(); throw new AMQPProtocolChannelException($reply_code, $reply_text, array($class_id, $method_id)); } /** * confirm a channel close */ protected function channel_close_ok($args) { $this->do_close(); } /** * enable/disable flow from peer */ public function flow($active) { list($class_id, $method_id, $args) = $this->protocolWriter->channelFlow($active); $this->send_method_frame(array($class_id, $method_id), $args); return $this->wait(array( $this->waitHelper->get_wait('channel.flow_ok') )); } protected function channel_flow($args) { $this->active = $args->read_bit(); $this->x_flow_ok($this->active); } protected function x_flow_ok($active) { list($class_id, $method_id, $args) = $this->protocolWriter->channelFlow($active); $this->send_method_frame(array($class_id, $method_id), $args); } protected function channel_flow_ok($args) { return $args->read_bit(); } protected function x_open($out_of_band="") { if ($this->is_open) { return; } list($class_id, $method_id, $args) = $this->protocolWriter->channelOpen($out_of_band); $this->send_method_frame(array($class_id, $method_id), $args); return $this->wait(array( $this->waitHelper->get_wait('channel.open_ok') )); } protected function channel_open_ok($args) { $this->is_open = true; if ($this->debug) { MiscHelper::debug_msg("Channel open"); } } /** * request an access ticket */ public function access_request($realm, $exclusive=false, $passive=false, $active=false, $write=false, $read=false) { list($class_id, $method_id, $args) = $this->protocolWriter->accessRequest($realm, $exclusive, $passive, $active, $write, $read); $this->send_method_frame(array($class_id, $method_id), $args); return $this->wait(array( $this->waitHelper->get_wait('access.request_ok') )); } /** * grant access to server resources */ protected function access_request_ok($args) { $this->default_ticket = $args->read_short(); return $this->default_ticket; } /** * declare exchange, create if needed */ public function exchange_declare($exchange, $type, $passive=false, $durable=false, $auto_delete=true, $internal=false, $nowait=false, $arguments=null, $ticket=null) { $arguments = $this->getArguments($arguments); $ticket = $this->getTicket($ticket); list($class_id, $method_id, $args) = $this->protocolWriter->exchangeDeclare( $ticket, $exchange, $type, $passive, $durable, $auto_delete, $internal, $nowait, $arguments ); $this->send_method_frame(array($class_id, $method_id), $args); if (!$nowait) { return $this->wait(array( $this->waitHelper->get_wait('exchange.declare_ok') )); } } /** * confirms an exchange declaration */ protected function exchange_declare_ok($args) { } /** * delete an exchange */ public function exchange_delete($exchange, $if_unused=false, $nowait=false, $ticket=null) { $ticket = $this->getTicket($ticket); list($class_id, $method_id, $args) = $this->protocolWriter->exchangeDelete($ticket, $exchange, $if_unused, $nowait); $this->send_method_frame(array($class_id, $method_id), $args); if (!$nowait) { return $this->wait(array( $this->waitHelper->get_wait('exchange.delete_ok') )); } } /** * confirm deletion of an exchange */ protected function exchange_delete_ok($args) { } /** * bind dest exchange to source exchange */ public function exchange_bind($destination, $source, $routing_key="", $nowait=false, $arguments=null, $ticket=null) { $arguments =