Producer configurable Keepalive Connection

Author: ikilobyte

src/Client.php

@@ -165,9 +165,12 @@ protected function initialization()
             }
 
             $this->topicManage->setConnection($topic, $this->connections[ $connectionKey ]);
+        }
 
-            // Add to EventLoop
-            $this->eventloop->addRead($this->topicManage->getConnection($topic));
+        // Add to Eventloop
+        foreach ($this->connections as $connection) {
+            $connection->setKeepalive($this->options->getKeepalive());
+            $this->eventloop->addRead($connection);
         }
 
         $this->isHandshake = true;

src/ConsumerOptions.php

@@ -16,7 +16,7 @@
  *
  * @package Pulsar
  */
-class ConsumerOptions extends Options
+final class ConsumerOptions extends Options
 {
     /**
      * consumer name

src/IO/AbstractIO.php

@@ -72,6 +72,16 @@ abstract class AbstractIO
     protected $lastSendPingCommandTime = 0;
 
 
+    /**
+     * @var int
+     */
+    protected $fd = 0;
+
+    /**
+     * @var bool
+     */
+    protected $keepalive = false;
+
     /**
      * @return void
      */
@@ -123,6 +133,33 @@ public function getMaxMessageSize()
     }
 
 
+
+    /**
+     * @param bool $status
+     * @return void
+     */
+    public function setKeepalive(bool $status)
+    {
+        $this->keepalive = $status;
+    }
+
+
+    /**
+     * @return int
+     */
+    public function fd(): int
+    {
+        return $this->fd;
+    }
+
+    /**
+     * Handling readable events
+     *
+     * @return mixed
+     */
+    abstract public function handleRead();
+
+    
     /**
      * @param string $host
      * @param int $port

src/IO/ChannelManager.php

@@ -0,0 +1,47 @@
+<?php
+/**
+ * Created by PhpStorm
+ * User: Sunny、
+ * Date: 2023/5/30
+ * Time: 16:54
+ */
+
+namespace Pulsar\IO;
+
+
+use Swoole\Coroutine\Channel;
+
+/**
+ * Class ChannelManager
+ *
+ * @package Pulsar\IO
+ */
+final class ChannelManager
+{
+
+    /**
+     * @var array
+     */
+    private static $inner = [];
+
+
+    /**
+     * @param int $fd
+     * @param int $size
+     * @return void
+     */
+    public static function init(int $fd, int $size = 10)
+    {
+        self::$inner[ $fd ] = new Channel($size);
+    }
+
+
+    /**
+     * @param int $fd
+     * @return Channel
+     */
+    public static function get(int $fd): Channel
+    {
+        return self::$inner[ $fd ];
+    }
+}

src/IO/StreamIO.php

@@ -57,6 +57,7 @@ public function connect(string $host, int $port, $timeout = null)
 
         $this->host = $host;
         $this->port = $port;
+        $this->fd = (int)$this->socket;
     }
 
 
@@ -173,6 +174,11 @@ public function close()
      */
     public function wait($seconds = null)
     {
+        // Receive from Channel
+        if ($this->keepalive) {
+            return ChannelManager::get($this->fd)->pop($seconds);
+        }
+
         $reads = [$this->socket];
         $n = stream_select($reads, $writes, $excepts, $seconds);
         if ($n <= 0) {
@@ -255,6 +261,6 @@ public function handleRead()
             } while ($remainingSize > 0);
         }
 
-        return new Response($buffer, $baseCommand);
+        return new Response($buffer, $baseCommand, $this->fd);
     }
 }

src/Options.php

@@ -129,6 +129,18 @@ public function getTopics(): array
     }
 
 
+
+    /**
+     * This connection pool is not enabled by default
+     *
+     * @return bool
+     */
+    public function getKeepalive(): bool
+    {
+        return false;
+    }
+
+    
     /**
      * @param ISchema $schema
      * @return void

src/Producer.php

@@ -12,6 +12,7 @@
 use Protobuf\AbstractMessage;
 use Pulsar\Exception\OptionsException;
 use Pulsar\Exception\RuntimeException;
+use Pulsar\IO\ChannelManager;
 use Pulsar\Proto\CommandSendReceipt;
 use Pulsar\Proto\KeyValue;
 use Pulsar\Proto\MessageMetadata;
@@ -20,6 +21,7 @@
 use Pulsar\Traits\ProducerKeepAlive;
 use Pulsar\Util\Buffer;
 use Pulsar\Util\Helper;
+use Swoole\Coroutine;
 
 /**
  * Class Producer
@@ -53,9 +55,22 @@ class Producer extends Client
      * @param string $url
      * @param ProducerOptions $options
      * @throws Exception\OptionsException
+     * @throws RuntimeException
      */
     public function __construct(string $url, ProducerOptions $options)
     {
+        // validate keepalive condition
+        if ($options->getKeepalive()) {
+
+            if (!extension_loaded('swoole')) {
+                throw new RuntimeException('Keepalive require swoole extension');
+            }
+
+            if (Coroutine::getCid() === -1) {
+                throw new RuntimeException('Keepalive Must be in a coroutine environment');
+            }
+        }
+
         parent::__construct($url, $options);
     }
 
@@ -66,12 +81,41 @@ public function __construct(string $url, ProducerOptions $options)
      */
     public function connect()
     {
+        // Establish tcp connection And complete the pulsar server handshake
         parent::initialization();
 
+        // Enable Keepalive
+        if ($this->options->getKeepalive()) {
+            Coroutine::create(function () {
+                while ($this->keepalive) {
+
+                    /**
+                     * @var $response Response
+                     */
+                    $response = $this->eventloop->wait(3);
+                    if (is_null($response)) {
+                        continue;
+                    }
+
+                    $fd = $response->fd();
+                    if ($fd <= 0) {
+                        continue;
+                    }
+
+                    // Push data to Channel
+                    ChannelManager::get($fd)->push($response);
+                }
+            });
+        }
+
         // Send CreateProducer Command
         foreach ($this->topicManage->all() as $id => $topic) {
-            $io = $this->topicManage->getConnection($topic);
-            $this->producers[] = new PartitionProducer($id, $topic, $io, $this->options);
+            $connection = $this->topicManage->getConnection($topic);
+            if ($this->options->getKeepalive()) {
+                ChannelManager::init($connection->fd());
+            }
+            
+            $this->producers[] = new PartitionProducer($id, $topic, $connection, $this->options);
         }
     }
 
@@ -123,6 +167,7 @@ public function send($payload, array $options = []): string
      * @return void
      * @throws RuntimeException|OptionsException
      * @throws \Exception
+     * @deprecated 1.3.0
      */
     public function sendAsync(string $payload, callable $callable, array $options = [])
     {
@@ -140,6 +185,7 @@ public function sendAsync(string $payload, callable $callable, array $options =
      * @return void
      * @throws Exception\IOException
      * @throws RuntimeException
+     * @deprecated 1.3.0
      */
     public function wait()
     {

src/ProducerOptions.php

@@ -17,7 +17,7 @@
  *
  * @package Pulsar
  */
-class ProducerOptions extends Options
+final class ProducerOptions extends Options
 {
     /**
      * @var string
@@ -35,6 +35,13 @@ class ProducerOptions extends Options
      */
     const INITIAL_SUBSCRIPTION_NAME = 'initial_subscription_name';
 
+
+    /**
+     * @var string
+     */
+    const KEEPALIVE = 'keepalive';
+
+
     /**
      * @param string $name
      * @return void
@@ -56,6 +63,17 @@ public function setInitialSubscriptionName(string $name)
         $this->data[ self::INITIAL_SUBSCRIPTION_NAME ] = $name;
     }
 
+    /**
+     * Whether to enable keep-alive
+     *
+     * @param bool $status
+     * @return void
+     */
+    public function setKeepalive(bool $status)
+    {
+        $this->data[ self::KEEPALIVE ] = $status;
+    }
+
 
     /**
      * @param int $type
@@ -99,4 +117,13 @@ public function getInitialSubscriptionName()
     {
         return $this->data[ self::INITIAL_SUBSCRIPTION_NAME ] ?? '';
     }
+
+
+    /**
+     * @return bool
+     */
+    public function getKeepalive(): bool
+    {
+        return $this->data[ self::KEEPALIVE ] ?? false;
+    }
 }

src/Response.php

@@ -40,16 +40,23 @@ class Response
     public $subCommand;
 
 
+    /**
+     * @var int
+     */
+    public $fd = 0;
+
     /**
      * @param Buffer $buffer
      * @param BaseCommand $message
+     * @param int $fd
      * @throws RuntimeException
      */
-    public function __construct(Buffer $buffer, BaseCommand $message)
+    public function __construct(Buffer $buffer, BaseCommand $message, int $fd)
     {
         $this->buffer = $buffer;
         $this->baseCommand = $message;
         $this->subCommand = $this->getSubCommand();
+        $this->fd = $fd;
         $this->checkError();
     }
 
@@ -71,6 +78,14 @@ public function getBuffer(): Buffer
         return $this->buffer;
     }
 
+    /**
+     * @return int
+     */
+    public function fd(): int
+    {
+        return $this->fd;
+    }
+
 
     /**
      * @return AbstractMessage

src/TopicManage.php

@@ -75,7 +75,7 @@ public function countPartitions(): int
     /**
      * @return int
      */
-    public function countTopics(): int
+    public function size(): int
     {
         return count($this->data);
     }