keepalive example

Author: ikilobyte

README-CN.md

@@ -73,16 +73,6 @@ for ($i = 0; $i < 10; $i++) {
     echo 'messageID ' . $messageID . "\n";
 }
 
-// Sending messages asynchronously
-//for ($i = 0; $i < 10; $i++) {
-//    $producer->sendAsync(sprintf('hello-async %d',$i),function(string $messageID){
-//        echo 'messageID ' . $messageID . "\n";
-//    });
-//}
-//
-//// Add this line when sending asynchronously
-//$producer->wait();
-
 // Sending delayed messages
 for ($i = 0; $i < 10; $i++) {
     $producer->send(sprintf('hello-delay %d',$i),[
@@ -93,10 +83,18 @@ for ($i = 0; $i < 10; $i++) {
 // close
 $producer->close();
 
-// or 
-// keepalive connection This method allows you to wrap the connection pool yourself
-// When the call is close() Will close the hold connection
-$producer->keepalive();
+```
+
+> 保持连接
+
+* 依赖 `Swoole` 扩展，必须开启协程
+* 如果是常驻内存应用，建议开启
+* 会保持连接，无需反复建立连接
+* 可以通过 `$producer->close()` 关闭连接，但不应该去调用
+* 请看[示例](./examples/producer-keepalive.php)
+
+```php
+$options->setKeepalive(true);
 ```
 
 > 消息去重

README.md

@@ -74,16 +74,6 @@ for ($i = 0; $i < 10; $i++) {
     echo 'messageID ' . $messageID . "\n";
 }
 
-// Sending messages asynchronously
-//for ($i = 0; $i < 10; $i++) {
-//    $producer->sendAsync(sprintf('hello-async %d',$i),function(string $messageID){
-//        echo 'messageID ' . $messageID . "\n";
-//    });
-//}
-//
-//// Add this line when sending asynchronously
-//$producer->wait();
-
 // Sending delayed messages
 for ($i = 0; $i < 10; $i++) {
     $producer->send(sprintf('hello-delay %d',$i),[
@@ -93,11 +83,18 @@ for ($i = 0; $i < 10; $i++) {
 
 // close
 $producer->close();
+```
+
+> Keepalive Connection (Recommended)
 
-// or 
-// keepalive connection This method allows you to wrap the connection pool yourself
-// When the call is close() Will close the hold connection
-$producer->keepalive();
+* require `Swoole` extension
+* If it is a resident memory application, it is recommended to open it.
+* Will keep connected, no need to repeatedly establish a connection
+* Calling the `close` method closes the connection
+* Please see [example](./examples/producer-keepalive.php)
+
+```php
+$options->setKeepalive(true);
 ```
 
 > Message deduplication
@@ -378,6 +375,7 @@ $reader->close();
     * setProducerName()
     * setCompression()
     * setSchema()
+    * setKeepalive()
 * ConsumerOptions
     * setTopic()
     * setTopics()

examples/producer-keepalive.php

@@ -0,0 +1,86 @@
+<?php
+
+use Pulsar\Compression\Compression;
+use Pulsar\Exception\IOException;
+use Pulsar\Exception\OptionsException;
+use Pulsar\Producer;
+use Pulsar\ProducerOptions;
+use Swoole\Http\Response;
+use Swoole\Http\Server;
+
+require_once __DIR__ . '/../vendor/autoload.php';
+
+
+
+/**
+ * Class ProducerStore
+ */
+class ProducerStore
+{
+
+    /**
+     * @var array<string,Producer>
+     */
+    protected static $inner = [];
+
+
+    /**
+     * @param string $topic
+     * @return Producer
+     * @throws IOException
+     * @throws OptionsException
+     * @throws \Pulsar\Exception\RuntimeException
+     */
+    public static function get(string $topic): Producer
+    {
+        if (!isset(self::$inner[ $topic ])) {
+            self::create($topic);
+        }
+
+        return self::$inner[ $topic ];
+    }
+
+
+    /**
+     * @param string $topic
+     * @return void
+     * @throws IOException
+     * @throws OptionsException
+     * @throws \Pulsar\Exception\RuntimeException
+     */
+    private static function create(string $topic)
+    {
+        $options = new ProducerOptions();
+
+        // If permission authentication is available
+        // Only JWT authentication is currently supported
+        // $options->setAuthentication(new Jwt('token'));
+
+        $options->setConnectTimeout(3);
+        $options->setTopic($topic);
+        $options->setCompression(Compression::ZLIB);
+        $options->setKeepalive(true);
+        $producer = new Producer('pulsar://localhost:6650', $options);
+        $producer->connect();
+
+        self::$inner[ $topic ] = $producer;
+    }
+}
+
+
+$server = new Server('0.0.0.0', 1234);
+$server->set([
+    'enable_coroutine' => true,
+    'hook_flags'       => SWOOLE_HOOK_ALL,
+]);
+
+$server->on('request', function ($req, Response $resp) {
+
+    // Should be taken from here to keep this connection from being closed
+    $producer = ProducerStore::get('persistent://public/default/demo');
+
+    $id = $producer->send('hello');
+    $resp->end(json_encode(['id' => $id]));
+});
+
+$server->start();