@@ -295,7 +295,7 @@ discard block |
||
| 295 | 295 | false, $this->getAutoAck(), |
| 296 | 296 | false, |
| 297 | 297 | false, |
| 298 | - function ($message) use ($processMessage) { |
|
| 298 | + function($message) use ($processMessage) { |
|
| 299 | 299 | /** @var AMQPMessage $message */ |
| 300 | 300 | $message = new Message($message); |
| 301 | 301 | return $processMessage($message); |
@@ -320,7 +320,7 @@ discard block |
||
| 320 | 320 | * @param AMQPChannel $channel |
| 321 | 321 | * @param AbstractConnection $connection |
| 322 | 322 | */ |
| 323 | - return function ($channel, $connection) { |
|
| 323 | + return function($channel, $connection) { |
|
| 324 | 324 | $channel->close(); |
| 325 | 325 | $connection->close(); |
| 326 | 326 | }; |
@@ -108,7 +108,9 @@ discard block |
||
| 108 | 108 | public function makeConnection(string $name) |
| 109 | 109 | { |
| 110 | 110 | $config = $this->configuration($name); |
| 111 | - if (isset($config['host'])) $config = [$config]; |
|
| 111 | + if (isset($config['host'])) { |
|
| 112 | + $config = [$config]; |
|
| 113 | + } |
|
| 112 | 114 | return AMQPStreamConnection::create_connection($config); |
| 113 | 115 | } |
| 114 | 116 | |
@@ -361,10 +363,15 @@ discard block |
||
| 361 | 363 | |
| 362 | 364 | $channel->queue_bind($this->getQueue(), $this->getExchange()); |
| 363 | 365 | |
| 364 | - if ($message instanceof Message) $message->getMessage(); |
|
| 366 | + if ($message instanceof Message) { |
|
| 367 | + $message->getMessage(); |
|
| 368 | + } |
|
| 365 | 369 | if (!$message instanceof AMQPMessage) { |
| 366 | - if (is_array($message) || is_object($message)) $message = json_encode($message); |
|
| 367 | - elseif ($message instanceof Collection) $message = $message->toJson(); |
|
| 370 | + if (is_array($message) || is_object($message)) { |
|
| 371 | + $message = json_encode($message); |
|
| 372 | + } elseif ($message instanceof Collection) { |
|
| 373 | + $message = $message->toJson(); |
|
| 374 | + } |
|
| 368 | 375 | $message = new AMQPMessage($message, ['content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT]); |
| 369 | 376 | } |
| 370 | 377 | |
@@ -23,7 +23,7 @@ |
||
| 23 | 23 | $this->mergeConfigFrom( |
| 24 | 24 | dirname(__DIR__) . '/../config/amqp.php', 'amqp' |
| 25 | 25 | ); |
| 26 | - $this->app->singleton(AMQPManager::class, function ($app) { |
|
| 26 | + $this->app->singleton(AMQPManager::class, function($app) { |
|
| 27 | 27 | return new AMQPManager($app); |
| 28 | 28 | }); |
| 29 | 29 | $this->app->alias(AMQPManager::class, 'amqp'); |
@@ -51,7 +51,7 @@ |
||
| 51 | 51 | * |
| 52 | 52 | * @return AMQPMessage |
| 53 | 53 | */ |
| 54 | - public function getMessage(){ |
|
| 54 | + public function getMessage() { |
|
| 55 | 55 | return $this->message; |
| 56 | 56 | } |
| 57 | 57 | |
@@ -40,11 +40,11 @@ |
||
| 40 | 40 | ->setQueue($this->queue) |
| 41 | 41 | ->setConsumerTag($this->consumerTag) |
| 42 | 42 | ->setAutoAck($this->autoAsk) |
| 43 | - ->consume(function ($message) { |
|
| 43 | + ->consume(function($message) { |
|
| 44 | 44 | /** @var Message $message */ |
| 45 | 45 | return static::processMessage($message); |
| 46 | 46 | }); |
| 47 | - } catch (InvalidArgumentException|ErrorException|Exception $exception) { |
|
| 47 | + } catch (InvalidArgumentException | ErrorException | Exception $exception) { |
|
| 48 | 48 | Log::error('AMQPMessage consume error:' . $exception->getMessage()); |
| 49 | 49 | } |
| 50 | 50 | } |