/**
  * Returns an AMQPConnection instance, ensuring that it is connected
  * @return \AMQPConnection
  */
 private function getConnection()
 {
     if (!$this->connection->isConnected()) {
         $this->connection->connect();
     }
     return $this->connection;
 }

 /**
  * Send an activity notice using AMQP
  * @param ActivityNotice $notice
  * @return bool
  */
 public function sendActivityNotice($notice)
 {
     if (!isset($notice)) {
         return false;
     }
     /** @var array $setting */
     $setting = $this->params['amqpSetting'];
     try {
         if ($this->amqpClientLibrary == "PhpAmqpLib") {
             $connection = new AMQPStreamConnection($setting['host'], $setting['port'], $setting['user'], $setting['password']);
             $channel = $connection->channel();
             $msg = new AMQPMessage(JsonHelper::encode($notice));
             $channel->basic_publish($msg, $setting['exchangeName'], $setting['routingKey']);
             $channel->close();
             $connection->close();
         } elseif ($this->amqpClientLibrary == "PECL") {
             $connection = new \AMQPConnection(['host' => $setting['host'], 'port' => $setting['port'], 'login' => $setting['user'], 'password' => $setting['password']]);
             $connection->connect();
             if ($connection->isConnected()) {
                 $channel = new \AMQPChannel($connection);
                 $exchange = new \AMQPExchange($channel);
                 $exchange->setName($setting['exchangeName']);
                 $exchange->publish(JsonHelper::encode($notice), $setting['routingKey']);
                 $connection->disconnect();
             }
         } else {
             return false;
         }
     } catch (\Exception $e) {
         return false;
     }
     return true;
 }

 public function it_should_give_connection_status(\AMQPConnection $connection)
 {
     $connection->isConnected()->willReturn(true);
     $this->isConnected()->shouldReturn(true);
     $connection->isConnected()->willReturn(false);
     $this->isConnected()->shouldReturn(false);
 }

 public function resolve(Command $command, InputInterface $input, OutputInterface $output, array $args)
 {
     $options = array_merge(array('host' => 'localhost', 'port' => 5763, 'login' => null, 'password' => null), $args);
     $conn = new \AMQPConnection($options);
     $conn->connect();
     $channel = new \AMQPChannel($conn);
     return new AmqpHandler(new \AMQPExchange($channel), $this->replacePlaceholders($args['name']));
 }

 protected function getAMQPExchange()
 {
     $connection = new \AMQPConnection(array('vhost' => 'swarrot'));
     $connection->connect();
     $channel = new \AMQPChannel($connection);
     $exchange = new \AMQPExchange($channel);
     $exchange->setName('exchange');
     return $exchange;
 }

 /**
  * Post a task to exchange specified in $details
  * @param AMQPConnection $connection Connection object
  * @param array $details Array of connection details
  * @param string $task JSON-encoded task
  * @param array $params AMQP message parameters
  */
 function PostToExchange($connection, $details, $task, $params)
 {
     $ch = new AMQPChannel($connection);
     $xchg = new AMQPExchange($ch);
     $xchg->setName($details['exchange']);
     $success = $xchg->publish($task, $details['binding'], 0, $params);
     $connection->disconnect();
     return $success;
 }

 protected function getAMQPQueue($name)
 {
     $connection = new \AMQPConnection(array('vhost' => 'swarrot'));
     $connection->connect();
     $channel = new \AMQPChannel($connection);
     $queue = new \AMQPQueue($channel);
     $queue->setName($name);
     return $queue;
 }

 /**
  * Connect to server
  * @return AMQPConnection
  */
 protected function _getConnection()
 {
     if (!$this->_connection) {
         $this->_connection = new \AMQPConnection($this->_config);
         if (!$this->_connection->connect()) {
             throw new \Exception("Cannot connect to the broker \n");
         }
     }
     return $this->_connection;
 }

 public function main()
 {
     $connection = new \AMQPConnection($this->config->get('mq'));
     try {
         $connection->connect();
         if (!$connection->isConnected()) {
             $this->logging->exception("Cannot connect to the broker!" . PHP_EOL);
         }
         $this->channel = new \AMQPChannel($connection);
         $this->exchange = new \AMQPExchange($this->channel);
         $this->exchange->setName($this->exchangeName);
         $this->exchange->setType(AMQP_EX_TYPE_DIRECT);
         //direct类型
         $this->exchange->setFlags(AMQP_DURABLE);
         //持久化
         $this->exchange->declareExchange();
         //echo "Exchange Status:".$this->exchange->declare()."\n";
         //创建队列
         $this->queue = new \AMQPQueue($this->channel);
         $this->queue->setName($this->queueName);
         $this->queue->setFlags(AMQP_DURABLE);
         //持久化
         $this->queue->declareQueue();
         $bind = $this->queue->bind($this->exchangeName, $this->routeKey);
         //echo "Message Total:".$this->queue->declare()."\n";
         //绑定交换机与队列，并指定路由键
         //echo 'Queue Bind: '.$bind."\n";
         //阻塞模式接收消息
         //while(true){
         //for($i=0; $i< self::loop ;$i++){
         //$this->queue->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
         $this->queue->consume(function ($envelope, $queue) {
             //print_r($envelope);
             //print_r($queue);
             $speed = microtime(true);
             $msg = $envelope->getBody();
             $result = $this->loader($msg);
             $queue->ack($envelope->getDeliveryTag());
             //手动发送ACK应答
             //$this->logging->info(''.$msg.' '.$result)
             $this->logging->debug('Protocol: ' . $msg . ' ');
             $this->logging->debug('Result: ' . $result . ' ');
             $this->logging->debug('Time: ' . (microtime(true) - $speed) . '');
         });
         $this->channel->qos(0, 1);
         //echo "Message Total:".$this->queue->declare()."\n";
         //}
     } catch (\AMQPConnectionException $e) {
         $this->logging->exception($e->__toString());
     } catch (\Exception $e) {
         $this->logging->exception($e->__toString());
         $connection->disconnect();
     }
 }

 /**
  * @param string $protocol
  * @param string $encoding
  * @param bool   $synchronous
  * @param array  $endpoint
  */
 public function __construct($protocol, $encoding, $synchronous = false, array $endpoint = [])
 {
     parent::__construct($protocol, $encoding, $synchronous, $endpoint);
     list($exchangeName, $routingKey) = array_values($this->endpoint);
     $credentials = array_filter($this->endpoint);
     $this->connection = new \AMQPConnection($credentials);
     $this->connection->connect();
     $this->channel = new \AMQPChannel($this->connection);
     $this->exchange = new \AMQPExchange($this->channel);
     $this->exchange->setName($exchangeName);
 }

 private static function sendMessage($exchange, $data)
 {
     global $CC_CONFIG;
     $conn = new AMQPConnection($CC_CONFIG["rabbitmq"]["host"], $CC_CONFIG["rabbitmq"]["port"], $CC_CONFIG["rabbitmq"]["user"], $CC_CONFIG["rabbitmq"]["password"], $CC_CONFIG["rabbitmq"]["vhost"]);
     $channel = $conn->channel();
     $channel->access_request($CC_CONFIG["rabbitmq"]["vhost"], false, false, true, true);
     $channel->exchange_declare($exchange, 'direct', false, true);
     $msg = new AMQPMessage($data, array('content_type' => 'text/plain'));
     $channel->basic_publish($msg, $exchange);
     $channel->close();
     $conn->close();
 }

 /**
  * Retrieves connection params from config, then inits connection and channel.
  *
  * @see http://www.php.net/manual/en/class.amqpconnection.php
  * @see http://www.php.net/manual/en/class.amqpchannel.php
  */
 public function __construct()
 {
     if (!extension_loaded('amqp')) {
         Mage::throwException('AMQP extension does not appear to be loaded');
     }
     $config = $this->getConfig();
     $this->_connection = new AMQPConnection($config);
     $this->_connection->connect();
     if (!$this->_connection->isConnected()) {
         Mage::throwException(sprintf("Unable to authenticate to 'amqp://%s:%d' (vhost: %s, user: %s, password: %s)", $config['host'], $config['port'], $config['vhost'], $config['user'], $config['password']));
     }
     $this->_channel = new AMQPChannel($this->_connection);
 }

 private static function sendMessage($exchange, $data)
 {
     $CC_CONFIG = Config::getConfig();
     $conn = new AMQPConnection($CC_CONFIG["rabbitmq"]["host"], $CC_CONFIG["rabbitmq"]["port"], $CC_CONFIG["rabbitmq"]["user"], $CC_CONFIG["rabbitmq"]["password"], $CC_CONFIG["rabbitmq"]["vhost"]);
     if (!isset($conn)) {
         throw new Exception("Cannot connect to RabbitMQ server");
     }
     $channel = $conn->channel();
     $channel->access_request($CC_CONFIG["rabbitmq"]["vhost"], false, false, true, true);
     $channel->exchange_declare($exchange, 'direct', false, true);
     $msg = new AMQPMessage($data, array('content_type' => 'text/plain'));
     $channel->basic_publish($msg, $exchange);
     $channel->close();
     $conn->close();
 }

 /**
  * {@inheritDoc}
  */
 public function createExchange()
 {
     // Create AMQP connection
     $amqpConnection = new \AMQPConnection(['host' => $this->host, 'port' => $this->port, 'vhost' => $this->vhost, 'login' => $this->login, 'password' => $this->password]);
     $amqpConnection->connect();
     // Create channel
     $channel = new \AMQPChannel($amqpConnection);
     // Create exchange
     $exchange = new \AMQPExchange($channel);
     $exchange->setName($this->exchangeName);
     $exchange->setType($this->exchangeType);
     $exchange->setFlags(AMQP_DURABLE);
     $exchange->declareExchange();
     return $exchange;
 }

 function __construct()
 {
     $connection = new \AMQPConnection();
     $connection->connect();
     if (!$connection->isConnected()) {
         throw new \AMQPConnectionException('Rabbit is not connected');
     }
     $this->channel = new \AMQPChannel($connection);
     $this->exchange = new \AMQPExchange($this->channel);
     //$this->exchange->delete('Celium');
     $this->exchange->setName('Celium');
     $this->exchange->setType('direct');
     //$this->exchange->setFlags(\AMQP_DURABLE);
     $this->exchange->declare();
 }

 /**
  * @return AMQPConnection
  */
 private function load_connection($name)
 {
     if (!class_exists('AMQPConnection')) {
         throw new Exception('MPF_AMQP_Factory Exception: AMQPConnection not found');
     }
     $config = MPF::get_instance()->get_config($name, self::CONF_F_AMQP);
     if (!isset($config['host'])) {
         throw new Exception('MPF_AMQP_Factory Exception: host undefined');
     }
     $con = new AMQPConnection($config);
     $rst = $con->connect();
     if ($rst == false) {
         throw new AMQPException('MPF_AMQP_Factory AMQPException: can not connect to server');
     }
     return $con;
 }

 public function __construct($AMQPConfiguration, $exchangeName = '')
 {
     if (!class_exists('AMQPConnection')) {
         throw new \LogicException("Can't find AMQP lib");
     }
     $ampq = new \AMQPConnection($AMQPConfiguration);
     if ($ampq->connect()) {
         $channel = new \AMQPChannel($ampq);
         $exchange = new \AMQPExchange($channel);
         $exchange->setName($exchangeName);
         $exchange->declare();
         $this->channel = $channel;
         $this->exchange = $exchange;
     } else {
         throw new \RuntimeException("Can't connect to AMQP server");
     }
 }

 function send($to, $subject, $text)
 {
     try {
         $conn = new AMQPConnection($this->rabbitMQConf['host'], $this->rabbitMQConf['port'], $this->rabbitMQConf['user'], $this->rabbitMQConf['pass']);
         $channel = $conn->channel();
         $channel->access_request($this->rabbitMQConf['virtualHost'], false, false, true, true);
         $mailData = array("to" => $to, "subject" => $subject, "text" => $text);
         $mailDataToJson = json_encode($mailData);
         $msg = new AMQPMessage($mailDataToJson, array('content_type' => 'text/plain'));
         $channel->basic_publish($msg, $this->queueConf['exchange']);
         $channel->close();
         $conn->close();
         return true;
     } catch (Exception $e) {
         echo "Something went wrong " . $e->getMessage();
     }
 }

 /**
  * @param Application $app
  */
 private function loadConnections(Application $app)
 {
     $app['amqp.connection'] = $app->share(function ($app) {
         if (!isset($app['amqp.connections'])) {
             throw new \InvalidArgumentException("You need to configure at least one AMQP connection");
         }
         $connections = [];
         foreach ($app['amqp.connections'] as $name => $options) {
             $connection = new \AMQPConnection($app['amqp.connections'][$name]);
             if (!$this->isLazy($app, $name)) {
                 $connection->connect();
             }
             $connections[$name] = $connection;
         }
         return $connections;
     });
 }

 /**
  * getChannel
  *
  * @param string $connectionName
  *
  * @return \AMQPChannel
  */
 public function getChannel($connectionName, $vhost)
 {
     $file = rtrim(getenv('HOME'), '/') . '/.rabbitmq_admin_toolkit';
     if (!file_exists($file)) {
         throw new \InvalidArgumentException('Can\'t find ~/.rabbitmq_admin_toolkit file');
     }
     $credentials = json_decode(file_get_contents($file), true);
     if (!isset($credentials[$connectionName])) {
         throw new \InvalidArgumentException("Connection {$connectionName} not found in ~/.rabbitmq_admin_toolkit");
     }
     $defaultCredentials = ['host' => '127.0.0.1', 'port' => 15672, 'user' => 'root', 'password' => 'root'];
     $credentials = array_merge($defaultCredentials, $credentials[$connectionName]);
     $credentials['login'] = $credentials['user'];
     unset($credentials['user'], $credentials['port']);
     $connection = new \AMQPConnection(array_merge($credentials, ['vhost' => $vhost]));
     $connection->connect();
     return new \AMQPChannel($connection);
 }