From 9d3f6d9ddd40248777587cd73da3c3da6f7111a8 Mon Sep 17 00:00:00 2001
From: Christopher Vagnetoft <cvagnetoft@gmail.com>
Date: Fri, 1 Mar 2024 19:10:47 +0100
Subject: [PATCH] Added a ConnectionException, misc fixes

---
 src/CommandBusClient.php    | 12 +++++++++---
 src/CommandBusException.php |  2 +-
 src/ConnectionException.php |  8 ++++++++
 src/MessageException.php    |  3 +--
 4 files changed, 19 insertions(+), 6 deletions(-)
 create mode 100644 src/ConnectionException.php

diff --git a/src/CommandBusClient.php b/src/CommandBusClient.php
index 7048c7c..c0509b6 100644
--- a/src/CommandBusClient.php
+++ b/src/CommandBusClient.php
@@ -32,6 +32,8 @@ class CommandBusClient implements CommandBusInterface
 
     private bool $autoReconnect = true;
 
+    private bool $isReconnecting = false;
+
     public function __construct(?CommandRegistry $commandRegistry = null, ?ConnectorInterface $connector = null)
     {
         $this->commandRegistry = $commandRegistry;
@@ -64,6 +66,7 @@ class CommandBusClient implements CommandBusInterface
             $this->connection = null;
         }
 
+        $this->isReconnecting = true;
         $this->connector->connect($this->address)->then(
             function (DuplexStreamInterface $connection) {
                 $this->connection = $connection;
@@ -73,12 +76,14 @@ class CommandBusClient implements CommandBusInterface
                     if ($this->autoReconnect) {
                         Loop::addTimer(1, $this->reconnect(...));
                     }
+                    $this->isReconnecting = false;
                 });
                 $connection->on('close', function () {
                     $this->emit(self::EVENT_DISCONNECTED);
                     if ($this->autoReconnect) {
                         Loop::addTimer(1, $this->reconnect(...));
                     }
+                    $this->isReconnecting = false;
                 });
                 $connection->on('data', function ($data) use ($connection) {
                     try {
@@ -142,10 +147,11 @@ class CommandBusClient implements CommandBusInterface
      */
     public function execute(string $command, array $context): PromiseInterface
     {
-        if (!$this->connection) {
+        if (!$this->connection || !$this->connection->isWritable()) {
+            $this->reconnect();
             return new Promise(function (callable $resolve, callable $canceller) {
-                $canceller(new \RuntimeException("Not connected to command bus."));
-            });
+                $canceller(new ConnectionException("Not connected to command bus."));
+            });    
         }
 
         $deferred = new Deferred();
diff --git a/src/CommandBusException.php b/src/CommandBusException.php
index f1baf9f..40e2888 100644
--- a/src/CommandBusException.php
+++ b/src/CommandBusException.php
@@ -2,7 +2,7 @@
 
 namespace NoccyLabs\React\CommandBus;
 
-interface CommandBusException
+class CommandBusException extends \RuntimeException
 {
 }
 
diff --git a/src/ConnectionException.php b/src/ConnectionException.php
new file mode 100644
index 0000000..4530216
--- /dev/null
+++ b/src/ConnectionException.php
@@ -0,0 +1,8 @@
+<?php
+
+namespace NoccyLabs\React\CommandBus;
+
+class ConnectionException extends CommandBusException
+{
+}
+
diff --git a/src/MessageException.php b/src/MessageException.php
index e68d504..99ff815 100644
--- a/src/MessageException.php
+++ b/src/MessageException.php
@@ -1,9 +1,8 @@
 <?php
 
 namespace NoccyLabs\React\CommandBus;
-use RuntimeException;
 
-class MessageException extends RuntimeException implements CommandBusException
+class MessageException extends CommandBusException
 {
 }