Add listeners for correct AMQP pool connections init, release and close.

.gitignore

@@ -1,2 +1,3 @@
 .idea
+.DS_Store
 composer.lock

src/Listener/CoroutineDeferListener.php

@@ -0,0 +1,31 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\Amqp\Listener;
+
+use Swoft\Amqp\Connection\ConnectionManager;
+use Swoft\Bean\BeanFactory;
+use Swoft\Event\Annotation\Mapping\Listener;
+use Swoft\Event\EventHandlerInterface;
+use Swoft\Event\EventInterface;
+use Swoft\SwoftEvent;
+
+/**
+ * Class CoroutineDeferListener
+ *
+ * @since 2.0
+ *
+ * @Listener(event=SwoftEvent::COROUTINE_DEFER)
+ */
+class CoroutineDeferListener implements EventHandlerInterface
+{
+    /**
+     * @param EventInterface $event
+     *
+     */
+    public function handle(EventInterface $event): void
+    {
+        /* @var ConnectionManager $conManager */
+        $conManager = BeanFactory::getBean(ConnectionManager::class);
+        $conManager->release();
+    }
+}

src/Listener/CoroutineDestoryListener.php

@@ -0,0 +1,31 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\Amqp\Listener;
+
+use Swoft\Amqp\Connection\ConnectionManager;
+use Swoft\Bean\BeanFactory;
+use Swoft\Event\Annotation\Mapping\Listener;
+use Swoft\Event\EventHandlerInterface;
+use Swoft\Event\EventInterface;
+use Swoft\SwoftEvent;
+
+/**
+ * Class CoroutineDestoryListener
+ *
+ * @since 2.0
+ *
+ * @Listener(event=SwoftEvent::COROUTINE_DESTROY)
+ */
+class CoroutineDestoryListener implements EventHandlerInterface
+{
+    /**
+     * @param EventInterface $event
+     *
+     */
+    public function handle(EventInterface $event): void
+    {
+        /* @var ConnectionManager $conManager */
+        $conManager = BeanFactory::getBean(ConnectionManager::class);
+        $conManager->release(true);
+    }
+}

src/Listener/WorkerStartListener.php

@@ -0,0 +1,36 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\Amqp\Listener;
+
+use Swoft\Amqp\Pool;
+use Swoft\Bean\BeanFactory;
+use Swoft\Connection\Pool\Exception\ConnectionPoolException;
+use Swoft\Event\Annotation\Mapping\Listener;
+use Swoft\Event\EventHandlerInterface;
+use Swoft\Event\EventInterface;
+use Swoft\Server\ServerEvent;
+
+/**
+ * Class WorkerStartListener
+ *
+ * @since 2.0
+ *
+ * @Listener(event=ServerEvent::WORK_PROCESS_START)
+ */
+class WorkerStartListener implements EventHandlerInterface
+{
+    /**
+     * @param EventInterface $event
+     *
+     * @throws ConnectionPoolException
+     */
+    public function handle(EventInterface $event): void
+    {
+        $pools = BeanFactory::getBeans(Pool::class);
+
+        /* @var Pool $pool */
+        foreach ($pools as $pool) {
+            $pool->initPool();
+        }
+    }
+}

src/Listener/WorkerStopAndErrorListener.php

@@ -0,0 +1,49 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\Amqp\Listener;
+
+use Swoft\Amqp\Pool;
+use Swoft\Bean\BeanFactory;
+use Swoft\Event\Annotation\Mapping\Subscriber;
+use Swoft\Event\EventInterface;
+use Swoft\Event\EventSubscriberInterface;
+use Swoft\Log\Helper\CLog;
+use Swoft\Server\SwooleEvent;
+use Swoft\SwoftEvent;
+
+/**
+ * Class WorkerStopAndErrorListener
+ *
+ * @since 2.0
+ *
+ * @Subscriber()
+ */
+class WorkerStopAndErrorListener implements EventSubscriberInterface
+{
+    /**
+     * @return array
+     */
+    public static function getSubscribedEvents(): array
+    {
+        return [
+            SwooleEvent::WORKER_STOP    => 'handle',
+            SwoftEvent::WORKER_SHUTDOWN => 'handle',
+        ];
+    }
+
+    /**
+     * @param EventInterface $event
+     *
+     */
+    public function handle(EventInterface $event): void
+    {
+        $pools = BeanFactory::getBeans(Pool::class);
+
+        /* @var Pool $pool */
+        foreach ($pools as $pool) {
+            $count = $pool->close();
+
+            CLog::info('Close %d AMQP connection on %s!', $count, $event->getName());
+        }
+    }
+}