Merge pull request #4 from versh23/add-emiter

add observers config
versh23 · Oct 14, 2019 · 0e7bbb1 · 0e7bbb1
2 parents 6df1959 + 886ea7c
commit 0e7bbb1
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 9 deletions.
diff --git a/.php_cs.cache b/.php_cs.cache
diff --git a/src/Messenger/Transport/Stomp/Connection.php b/src/Messenger/Transport/Stomp/Connection.php
@@ -13,6 +13,8 @@
 use Interop\Queue\Exception\InvalidDestinationException;
 use Interop\Queue\Exception\InvalidMessageException;
 use Stomp\Exception\ConnectionException;
+use Stomp\Network\Observer\HeartbeatEmitter;
+use Stomp\Network\Observer\ServerAliveObserver;
 
 class Connection
 {
@@ -33,6 +35,8 @@ public static function create(string $dsn, array $options = []): self
         $destinationName = $options['destination'] ?? null;
         $queueName = $options['queue'] ?? null;
         $topicName = $options['topic'] ?? null;
+        $useHeartbeatEmitter = $options['useHeartbeatEmitter'] ?? false;
+        $useServerAliveObserver = $options['useServerAliveObserver'] ?? false;
 
         if (!$destinationName && $queueName) {
             $destinationName = '/queue/'.$queueName;
@@ -67,8 +71,35 @@ public static function create(string $dsn, array $options = []): self
         }
 
         $factory = new StompConnectionFactory($config);
+        $context = $factory->createContext();
 
-        return new self($factory->createContext(), $destinationName);
+        if ($useHeartbeatEmitter || $useServerAliveObserver) {
+            $stomp = $context->getStomp();
+            $stomp->disconnect();
+            $connection = $stomp->getConnection();
+
+            $sendBeat = 0;
+            $receiveBeat = 0;
+
+            if ($useHeartbeatEmitter) {
+                $sendBeat = 3000;
+                $emitter = new HeartbeatEmitter($connection);
+                $connection->getObservers()->addObserver($emitter);
+            }
+
+            if ($useServerAliveObserver) {
+                $receiveBeat = 3000;
+                $connection->getObservers()->addObserver(new ServerAliveObserver());
+            }
+
+            $stomp->setHeartbeat($sendBeat, $receiveBeat);
+
+            $connection->setReadTimeout(0, 3 * 500000);
+
+            $stomp->connect();
+        }
+
+        return new self($context, $destinationName);
     }
 
     /**

diff --git a/src/Messenger/Transport/Stomp/StompReceiver.php b/src/Messenger/Transport/Stomp/StompReceiver.php
@@ -5,7 +5,6 @@
 namespace Versh23\Messenger\Transport\Stomp;
 
 use Enqueue\Stomp\StompMessage;
-use Stomp\Exception\ConnectionException;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Exception\TransportException;
@@ -49,12 +48,6 @@ public function __construct(Connection $connection, SerializerInterface $seriali
      */
     public function get(): iterable
     {
-        try {
-            $this->connection->ping();
-        } catch (ConnectionException $e) {
-            throw new TransportException($e->getMessage());
-        }
-
         $stompMessage = $this->connection->get();
 
         if (!$stompMessage) {