Merge pull request #1099 from atrauzzi/fix/dead-non-consumer-connections

Fix - Add automatic reconnect support for STOMP producers

.gitignore

@@ -9,8 +9,11 @@ bin/jp.php
 bin/php-parse
 bin/google-cloud-batch
 bin/thruway
+bin/phpstan.phar
+bin/var-dump-server
+bin/yaml-lint
 vendor
 var
 .php_cs
 .php_cs.cache
-composer.lock
+composer.lock

pkg/stomp/StompConnectionFactory.php

@@ -74,14 +74,14 @@ public function __construct($config = 'stomp:')
      */
     public function createContext(): Context
     {
-        if ($this->config['lazy']) {
-            return new StompContext(
-                function () { return $this->establishConnection(); },
-                $this->config['target']
-            );
-        }
+        $stomp = $this->config['lazy']
+            ? function () { return $this->establishConnection(); }
+        : $this->establishConnection();
+
+        $target = $this->config['target'];
+        $detectTransientConnections = (bool) $this->config['detect_transient_connections'];
 
-        return new StompContext($this->establishConnection(), $this->config['target']);
+        return new StompContext($stomp, $target, $detectTransientConnections);
     }
 
     private function establishConnection(): BufferedStompClient
@@ -169,6 +169,7 @@ private function defaultConfig(): array
             'read_timeout' => 60,
             'send_heartbeat' => 0,
             'receive_heartbeat' => 0,
+            'detect_transient_connections' => false,
         ];
     }
 }

pkg/stomp/StompContext.php

@@ -38,10 +38,15 @@ class StompContext implements Context
      */
     private $stompFactory;
 
+    /**
+     * @var bool
+     */
+    private $transient;
+
     /**
      * @param BufferedStompClient|callable $stomp
      */
-    public function __construct($stomp, string $extensionType)
+    public function __construct($stomp, string $extensionType, bool $detectTransientConnections = false)
     {
         if ($stomp instanceof BufferedStompClient) {
             $this->stomp = $stomp;
@@ -53,6 +58,7 @@ public function __construct($stomp, string $extensionType)
 
         $this->extensionType = $extensionType;
         $this->useExchangePrefix = ExtensionType::RABBITMQ === $extensionType;
+        $this->transient = $detectTransientConnections;
     }
 
     /**
@@ -173,6 +179,8 @@ public function createConsumer(Destination $destination): Consumer
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, StompDestination::class);
 
+        $this->transient = false;
+
         return new StompConsumer($this->getStomp(), $destination);
     }
 
@@ -181,6 +189,10 @@ public function createConsumer(Destination $destination): Consumer
      */
     public function createProducer(): Producer
     {
+        if ($this->transient && $this->stomp) {
+            $this->stomp->disconnect();
+        }
+
         return new StompProducer($this->getStomp());
     }
 
@@ -202,14 +214,20 @@ public function purgeQueue(Queue $queue): void
     public function getStomp(): BufferedStompClient
     {
         if (false == $this->stomp) {
-            $stomp = call_user_func($this->stompFactory);
-            if (false == $stomp instanceof BufferedStompClient) {
-                throw new \LogicException(sprintf('The factory must return instance of BufferedStompClient. It returns %s', is_object($stomp) ? get_class($stomp) : gettype($stomp)));
-            }
-
-            $this->stomp = $stomp;
+            $this->stomp = $this->createStomp();
         }
 
         return $this->stomp;
     }
+
+    private function createStomp(): BufferedStompClient
+    {
+        $stomp = call_user_func($this->stompFactory);
+
+        if (false == $stomp instanceof BufferedStompClient) {
+            throw new \LogicException(sprintf('The factory must return instance of BufferedStompClient. It returns %s', is_object($stomp) ? get_class($stomp) : gettype($stomp)));
+        }
+
+        return $stomp;
+    }
 }

pkg/stomp/Tests/StompConnectionFactoryConfigTest.php

@@ -70,6 +70,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -91,6 +92,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -112,6 +114,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -134,6 +137,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -156,6 +160,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -178,6 +183,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -201,6 +207,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -222,6 +229,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
 
@@ -244,6 +252,7 @@ public static function provideConfigs()
                 'read_timeout' => 60,
                 'send_heartbeat' => 0,
                 'receive_heartbeat' => 0,
+                'detect_transient_connections' => false,
             ],
         ];
     }

pkg/stomp/Tests/StompContextTest.php

@@ -144,7 +144,7 @@ public function testShouldCloseConnections()
     {
         $client = $this->createStompClientMock();
         $client
-            ->expects($this->once())
+            ->expects($this->atLeastOnce())
             ->method('disconnect')
         ;