Add exceptionCallback in CommonConfig

Author: Yurunsoft

doc/consumer.en.md

@@ -36,6 +36,7 @@ Class `longlang\phpkafka\Consumer\ConsumerConfig`
 | groupHeartbeat | Group heartbeat intervals. (unit: second) | `3` |
 | autoCreateTopic | Auto create topic. | `true` |
 | partitionAssignmentStrategy | Consumer partition assignment strategy. Optional: Range-`longlang\phpkafka\Consumer\Assignor\RangeAssignor`, RoundRobin-`\longlang\phpkafka\Consumer\Assignor\RoundRobinAssignor`, Sticky-`\longlang\phpkafka\Consumer\Assignor\StickyAssignor`. |
+| exceptionCallback | This callback is called when an exception that cannot be thrown by the `recv()` coroutine is encountered. Format: `function(\Exception $e){}` | `null` |
 
 ## Asynchronous (callback)
 

doc/consumer.md

@@ -36,6 +36,7 @@
 | groupHeartbeat | 分组心跳时间间隔，单位：秒 | `3` |
 | autoCreateTopic | 自动创建主题 | `true` |
 | partitionAssignmentStrategy | 消费者分区分配策略，可选：范围分配-`longlang\phpkafka\Consumer\Assignor\RangeAssignor`、轮询分配-`\longlang\phpkafka\Consumer\Assignor\RoundRobinAssignor`、粘性分配-`\longlang\phpkafka\Consumer\Assignor\StickyAssignor` | `longlang\phpkafka\Consumer\Assignor\RangeAssignor` |
+| exceptionCallback | 遇到无法在`recv()`协程抛出的异常时，调用此回调。格式：`function(\Exception $e){}` | `null` |
 
 ## 异步消费（回调）
 

doc/producer.en.md

@@ -25,6 +25,7 @@ Class `longlang\phpkafka\Producer\ProducerConfig`
 | producerEpoch | producer Epoch | `-1` |
 | partitionLeaderEpoch | partition Leader Epoch | `-1` |
 | autoCreateTopic | auto create topic | `true` |
+| exceptionCallback | This callback is called when an exception that cannot be thrown by the `recv()` coroutine is encountered. Format: `function(\Exception $e){}` | `null` |
 
 ## Send a single message
 

doc/producer.md

@@ -25,6 +25,7 @@
 | producerEpoch | 生产者 Epoch | `-1` |
 | partitionLeaderEpoch | 分区 Leader Epoch | `-1` |
 | autoCreateTopic | 自动创建主题 | `true` |
+| exceptionCallback | 遇到无法在`recv()`协程抛出的异常时，调用此回调。格式：`function(\Exception $e){}` | `null` |
 
 ## 发送单个消息
 

src/Client/SwooleClient.php

@@ -137,13 +137,17 @@ private function startRecvCo()
                     $length = Int32::unpack($data);
                     $data = $this->socket->recv($length);
                     $correlationId = Int32::unpack($data);
-                    if (!isset($this->recvChannels[$correlationId])) {
-                        continue;
+                    if (isset($this->recvChannels[$correlationId])) {
+                        $this->recvChannels[$correlationId]->push($data);
                     }
                 } catch (Exception $e) {
-                    $data = $e;
+                    $callback = $this->getConfig()->getExceptionCallback();
+                    if ($callback) {
+                        $callback($e);
+                    } else {
+                        throw $e;
+                    }
                 }
-                $this->recvChannels[$correlationId]->push($data);
             }
         });
     }

src/Config/CommonConfig.php

@@ -47,6 +47,13 @@ class CommonConfig extends AbstractConfig
      */
     protected $updateBrokers = true;
 
+    /**
+     * Exception callback in SwooleClient.
+     *
+     * @var callable|null
+     */
+    protected $exceptionCallback = null;
+
     public function getConnectTimeout(): float
     {
         return $this->connectTimeout;
@@ -152,4 +159,16 @@ public function setUpdateBrokers(bool $updateBrokers): self
 
         return $this;
     }
+
+    public function getExceptionCallback(): ?callable
+    {
+        return $this->exceptionCallback;
+    }
+
+    public function setExceptionCallback(?callable $exceptionCallback): self
+    {
+        $this->exceptionCallback = $exceptionCallback;
+
+        return $this;
+    }
 }

tests/Client/SwooleClientTest.php

@@ -4,6 +4,7 @@
 
 namespace longlang\phpkafka\Test\Client;
 
+use Exception;
 use longlang\phpkafka\Client\ClientInterface;
 use longlang\phpkafka\Client\SwooleClient;
 use longlang\phpkafka\Test\TestUtil;
@@ -63,4 +64,15 @@ public function testClose(ClientInterface $client)
 
         return parent::testClose($client);
     }
+
+    public function testExceptionCallback()
+    {
+        $client = $this->testClient();
+        $exception = null;
+        $client->getConfig()->setExceptionCallback(function (Exception $e) use (&$exception) {
+            $exception = $e;
+        });
+        $client->close();
+        $this->assertInstanceOf(Exception::class, $exception);
+    }
 }