modify

HttpServer/Server.php

@@ -14,13 +14,9 @@ use HttpServer\Service\Receive;
 use HttpServer\Service\Packet;
 use HttpServer\Service\Websocket;
 use Exception;
-use JetBrains\PhpStorm\Pure;
 use ReflectionException;
-use Rpc\Service;
 use Snowflake\Abstracts\Config;
 use Snowflake\Error\LoggerProcess;
-use Snowflake\Event;
-use Snowflake\Exception\ComponentException;
 use Snowflake\Exception\ConfigException;
 use Snowflake\Exception\NotFindClassException;
 use Snowflake\Process\Biomonitoring;
@@ -53,7 +49,6 @@ class Server extends HttpService
     const PACKAGE = 'PACKAGE';
     const WEBSOCKET = 'WEBSOCKET';
 
-    private array $listening = [];
     private array $server = [
         'HTTP'      => [SWOOLE_TCP, Http::class],
         'TCP'       => [SWOOLE_TCP, Receive::class],
@@ -225,10 +220,10 @@ class Server extends HttpService
      * @throws ConfigException
      * @throws Exception
      */
-    public function onProcessListener(): void
+    public function onProcessListener(): mixed
     {
         if (!($this->swoole instanceof \Swoole\Server)) {
-            return;
+            return $this->swoole;
         }
 
         $processes = Config::get('processes');
@@ -237,6 +232,7 @@ class Server extends HttpService
         } else {
             $this->deliveryProcess($this->process);
         }
+        return $this->swoole;
     }
 
 
@@ -359,12 +355,32 @@ class Server extends HttpService
         if (isset($config['settings']) && is_array($config['settings'])) {
             $newListener->set($config['settings']);
         }
-        $this->onListenerBind($config, $this->swoole);
+        $this->onListenerBind($config);
 
         return $this->swoole;
     }
 
 
+    /**
+     * @param $mode
+     * @return bool
+     */
+    public static function isTcp($mode)
+    {
+        return in_array($mode, [SWOOLE_SOCK_TCP, SWOOLE_TCP, SWOOLE_TCP6, SWOOLE_SOCK_TCP6]);
+    }
+
+
+    /**
+     * @param $mode
+     * @return bool
+     */
+    public static function isUdp($mode)
+    {
+        return in_array($mode, [SWOOLE_SOCK_UDP, SWOOLE_UDP, SWOOLE_UDP6, SWOOLE_SOCK_UDP6]);
+    }
+
+
     /**
      * @return Packet|Websocket|Receive|Http|null
      * @throws ConfigException
@@ -372,12 +388,11 @@ class Server extends HttpService
      */
     private function startRpcService(): Packet|Websocket|Receive|Http|null
     {
-        $rpcService = Config::get('rpc.enable', []);
+        $rpcService = Config::get('rpc', []);
-        if ($rpcService === true) {
+        if (empty($rpcService)) {
-            /** @var Service $service */
+            return $this->swoole;
-            $service = Snowflake::app()->get('rpc-service');
-            $service->instance($this->swoole);
         }
+        $this->addListener($rpcService);
         return $this->swoole;
     }
 
@@ -394,17 +409,13 @@ class Server extends HttpService
         if (is_array($config['settings'] ?? null)) {
             $settings = array_merge($settings, $config['settings']);
         }
-        $this->swoole = $this->createServer($class, $config);
-        $settings['daemonize'] = $this->daemon;
-        if (!isset($settings['pid_file'])) {
-            $settings['pid_file'] = PID_PATH;
-        }
-
         $this->debug(Snowflake::listen($config));
-        $this->swoole->set($settings);
+        $this->swoole = $this->createServer($class, $config);
-        $this->onProcessListener();
+        $this->swoole->set(array_merge($settings, [
-
+            'daemonize' => $this->daemon,
-        return $this->swoole;
+            'pid_file'  => $settings['pid_file'] ?? PID_PATH
+        ]));
+        return $this->onProcessListener();
     }
 
 
@@ -419,16 +430,6 @@ class Server extends HttpService
     }
 
 
-    /**
-     * @param $listen
-     * @return bool
-     */
-    #[Pure] public function isListen($listen): bool
-    {
-        return in_array($listen, $this->listenTypes);
-    }
-
-
     /**
      * @param $config
      * @param $newListener
@@ -437,16 +438,10 @@ class Server extends HttpService
      * @throws ReflectionException
      * @throws Exception
      */
-    private function onListenerBind($config, $newListener): Packet|Websocket|Receive|Http|null
+    private function onListenerBind($config): Packet|Websocket|Receive|Http|null
     {
-        if (!in_array($config['type'], [self::HTTP, self::TCP, self::PACKAGE])) {
+        $this->bindServerEvent($config['type']);
-            throw new Exception('Unknown server type(' . $config['type'] . ').');
+
-        }
-        if ($config['type'] == self::HTTP && !$this->swoole->getCallback('request')) {
-            $this->onBindCallback('request', [make(OnRequest::class), 'onHandler']);
-        } else {
-            $this->noHttp($newListener, $config);
-        }
         $this->debug(sprintf('Check listen %s::%d -> ok', $config['host'], $config['port']));
         $this->listenTypes[] = $config['type'];
         return $this->swoole;
@@ -458,14 +453,20 @@ class Server extends HttpService
      * @param $config
      * @throws Exception
      */
-    private function noHttp($newListener, $config)
+    private function bindServerEvent($type = self::TCP)
     {
+        if (in_array($type, [self::PACKAGE, self::TCP])) {
             $this->onBindCallback('connect', [make(OnConnect::class), 'onHandler']);
             $this->onBindCallback('close', [make(OnClose::class), 'onHandler']);
-        if ($config['type'] == self::TCP) {
+            if ($type == self::PACKAGE) {
-            $this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']);
-        } else {
                 $this->onBindCallback('packet', [make(OnPacket::class), 'onHandler']);
+            } else if ($type == self::TCP) {
+                $this->onBindCallback('receive', [make(OnReceive::class), 'onHandler']);
+            }
+        } else if ($type === self::HTTP) {
+            $this->onBindCallback('request', [make(OnRequest::class), 'onHandler']);
+        } else {
+            throw new Exception('Unknown server type(' . $type . ').');
         }
     }
 

Kafka/KafkaProviders.php

@@ -36,7 +36,7 @@ class KafkaProviders extends Providers
 			return;
 		}
 
-		$kafkaServers = Config::get('kafka.servers', []);
+		$kafkaServers = Config::get('kafka.consumers', []);
 		if (empty($kafkaServers)) {
 			return;
 		}

Kafka/Producer.php

@@ -36,6 +36,7 @@ class Producer extends Component
 
     private ?\RdKafka\Producer $producer = null;
 
+    private bool $isAck = true;
 
     /**
      * Producer constructor.
@@ -50,6 +51,9 @@ class Producer extends Component
         }
         $this->conf = new Conf();
         $this->topicConf = new TopicConf();
+        $this->conf->setErrorCb(function ($kafka, $err, $reason) {
+            $this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
+        });
     }
 
 
@@ -94,9 +98,62 @@ class Producer extends Component
      */
     public function dispatch(string $topic, array $params = [], string $groupId = null)
     {
-		$consumers = Config::get('kafka.consumers.' . $topic);
+        $this->beforePushMessage($topic, $groupId);
+        $this->sendMessage($topic, [$params]);
+    }
+
+    /**
+     * @return \RdKafka\Producer
+     * @throws Exception
+     */
+    private function getProducer(): \RdKafka\Producer
+    {
+        $pool = Snowflake::app()->getChannel();
+        return $pool->pop(\RdKafka\Producer::class, function () {
+            return Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
+        });
+    }
+
+
+    /**
+     * @return \RdKafka\Producer
+     * @throws Exception
+     */
+    private function getProducerTopic(\RdKafka\Producer $producer, $topic): ProducerTopic
+    {
+        $pool = Snowflake::app()->getChannel();
+        return $pool->pop($topic . '::' . ProducerTopic::class, function () use ($producer, $topic) {
+            return $producer->newTopic($topic, $this->topic_conf);
+        });
+    }
+
+
+    /**
+     * @param string $toPic
+     * @param string|null $key
+     * @param array $data
+     * @param string|null $groupId
+     */
+    public function batch(string $toPic, ?string $key, array $data, ?string $groupId = null)
+    {
+        $this->beforePushMessage($toPic, $groupId);
+
+        $this->sendMessage($toPic, $data, $key);
+    }
+
+
+    /**
+     * @param $topic
+     * @param $groupId
+     * @param $brokers
+     * @return ProducerTopic
+     * @throws \Snowflake\Exception\ConfigException
+     */
+    private function beforePushMessage($topic, $groupId): void
+    {
+        $consumers = Config::get('kafka.producers.' . $topic);
         if (empty($consumers) || !is_array($consumers)) {
-			return;
+            throw new Exception('You need set kafka.producers config');
         }
         if (!isset($consumers['brokers'])) {
             throw new Exception('You need set brokers config.');
@@ -106,82 +163,67 @@ class Producer extends Component
         } else if (!isset($consumers['groupId'])) {
             $consumers['groupId'] = $topic . ':' . Snowflake::localhost();
         }
-		$this->setGroupId($consumers['groupId'])->setTopic($topic)
+        $this->setGroupId($consumers['groupId']);
-			->setBrokers($consumers['brokers'])
+        $this->setBrokers($consumers['brokers']);
-			->delivery(swoole_serialize($params));
+        $this->setTopic($topic);
     }
 
 
     /**
-	 * @param $message
+     * @param TopicConf $topicConf
-	 * @param null $key
+     * @param string $topic
-	 * @param bool $isAck
+     * @param array $message
-	 * @throws NotFindClassException
+     * @param string $key
-	 * @throws ReflectionException
      * @throws Exception
      */
-	public function delivery($message, $key = null, $isAck = false)
+    private function sendMessage(string $topic, array $message, string $key = '')
     {
-		if (!$this->conf || !$this->topicConf) {
+        $producer = $this->getProducer();
-			throw new Exception('Error. Please set kafka conf.');
+        $producerTopic = $this->getProducerTopic($producer, $topic);
-		}
-		$this->conf->setErrorCb(function ($kafka, $err, $reason) {
-			$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
-		});
 
-		$event = Snowflake::app()->getEvent();
+        if ($this->isAck) {
-		$event->on(Event::SYSTEM_RESOURCE_RELEASES, [$this, 'flush']);
+            $this->flush($producer);
-
-		if ($this->producer === null) {
-			$this->producer = Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
         }
 
-		$this->setTopicAcks($isAck);
+        foreach ($message as $value) {
-		$this->push($message, $key);
+            $producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, $value, $key);
-		if ($isAck === true) {
+            $producer->poll(0);
-			$this->flush();
         }
-	}
+        $this->flush($producer);
-
+        $this->recover($topic, $producer, $producerTopic);
-
-	/** @var ProducerTopic[] $topics */
-	private array $topics = [];
-
-
-	/**
-	 * @param $message
-	 * @param $key
-	 */
-	private function push($message, $key)
-	{
-		if (!isset($this->topics[$this->_topic])) {
-			$this->topics[$this->_topic] = $this->producer->newTopic($this->_topic, $this->topicConf);
-		}
-		$this->topics[$this->_topic]->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
-		$this->producer->poll(0);
     }
 
 
     /**
-	 * @param $isAck
+     * @param bool $ack
      */
-	private function setTopicAcks(bool $isAck)
+    public function setAsk(bool $ack){
-	{
+        $this->isAck = $ack;
-		if ($isAck) {
+        $this->topicConf->set('request.required.acks', $this->isAck ? '1' : '0');
-			if ($this->producer->getOutQLen() > 0) {
-				$this->flush();
-			}
-			$this->topicConf->set('request.required.acks', '1');
-		} else {
-			$this->topicConf->set('request.required.acks', '0');
-		}
     }
 
 
-	public function flush()
+
+    /**
+     * @param \RdKafka\Producer $producer
+     * @param ProducerTopic $producerTopic
+     * @throws Exception
+     */
+    private function recover(string $topic, \RdKafka\Producer $producer, ProducerTopic $producerTopic)
     {
-		while ($this->producer->getOutQLen() > 0) {
+        $channel = Snowflake::app()->getChannel();
-			$result = $this->producer->flush(100);
+        $channel->push($producerTopic, $topic . '::' . ProducerTopic::class);
+        $channel->push($producer, \RdKafka\Producer::class);
+    }
+
+
+    /**
+     * @param \RdKafka\Producer $producer
+     */
+    public function flush(\RdKafka\Producer $producer)
+    {
+        while ($producer->getOutQLen() > 0) {
+            $result = $producer->flush(100);
             if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
                 break;
             }

function.php

@@ -588,6 +588,26 @@ if (!function_exists('listen')) {
 }
 
 
+if (!function_exists('event')) {
+
+
+    /**
+     * @param $name
+     * @param $callback
+     * @param $params
+     * @param $isAppend
+     * @throws Exception
+     * @throws Exception
+     */
+    function event($name, $callback, $params = [], $isAppend = true)
+    {
+        $event = Snowflake::app()->getEvent();
+        $event->on($name, $callback, $params, $isAppend);
+    }
+
+}
+
+
 if (!function_exists('alias')) {
 
     /**