[WIP]feature(server): integrate tcp server and listenerManager

bootstrap.php

@@ -0,0 +1,30 @@
+<?php
+
+use Asiries335\redisSteamPhp\Consumer\DemoConsumer;
+use Asiries335\redisSteamPhp\Consumer\Manager\TcpConsumerManager;
+use Asiries335\redisSteamPhp\Server\TcpServer;
+use Asiries335\redisSteamPhp\ServerProvider;
+
+require_once __DIR__ . '/vendor/autoload.php';
+
+// set server
+$tcpServer = new TcpServer();
+$tcpServer->setConfig([
+    'host' => '0.0.0.0',
+    'port' => '1236'
+]);
+
+// set listener manager
+$tcpConsumerManager = new TcpConsumerManager();
+$tcpConsumerManager->setConsumers([
+    new DemoConsumer()
+]);
+
+// run
+$serverProvider = new ServerProvider(
+    $tcpServer,
+    $tcpConsumerManager
+);
+
+$serverProvider->boot();
+

composer.json

@@ -11,7 +11,10 @@
     "require": {
         "predis/predis": "^1.1",
         "react/event-loop": "^1.1.1",
-        "dto/dto": "^3.2"
+        "dto/dto": "^3.2",
+        "react/stream": "^1.2",
+        "clue/redis-react": "^2.6",
+        "ext-sockets": "*"
     },
     "require-dev": {
         "phpunit/phpunit": "^8.5",

src/Consumer/ConsumerInterface.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Consumer;
+
+interface ConsumerInterface
+{
+    public function handle($payload);
+}

src/Consumer/DemoConsumer.php

@@ -0,0 +1,11 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Consumer;
+
+class DemoConsumer implements ConsumerInterface
+{
+
+    public function handle($payload) {
+        echo 'привет я Demo Consumer - ' .  $payload;
+    }
+}

src/Consumer/Manager/ConsumerManagerInterface.php

@@ -0,0 +1,26 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Consumer\Manager;
+
+use Asiries335\redisSteamPhp\Consumer\ConsumerInterface;
+
+interface ConsumerManagerInterface
+{
+    /**
+     * @param ConsumerInterface[] $consumers
+     * @return void
+     */
+    public function setConsumers(array $consumers): void;
+
+    /**
+     * @param mixed $payload
+     * @return void
+     */
+    public function handle($payload): void;
+
+
+    /**
+     * @return ConsumerInterface[]
+     */
+    public function list(): array;
+}

src/Consumer/Manager/TcpConsumerManager.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Consumer\Manager;
+
+use Asiries335\redisSteamPhp\Consumer\ConsumerInterface;
+
+class TcpConsumerManager implements ConsumerManagerInterface
+{
+    /**
+     * @var ConsumerInterface[]
+     */
+    private array $consumers;
+
+    /**
+     * @param ConsumerInterface[] $consumers
+     * @return void
+     */
+    public function setConsumers(array $consumers): void {
+       $this->consumers = $consumers;
+    }
+
+    /**
+     * @param mixed $payload
+     * @return void
+     */
+    public function handle($payload): void {
+        foreach ($this->consumers as $consumer) {
+            $consumer->handle($payload);
+        }
+    }
+
+    /**
+     * @return ConsumerInterface[]
+     */
+    public function list(): array {
+        return $this->consumers;
+    }
+}

src/Server/ServerInterface.php

@@ -0,0 +1,31 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Server;
+
+use Asiries335\redisSteamPhp\Consumer\ConsumerInterface;
+
+interface ServerInterface
+{
+
+    /**
+     * @param ConsumerInterface[] $consumers
+     * @return void
+     */
+    public function setConsumers(array $consumers): void;
+
+    /**
+     * @param array $config
+     * @return void
+     */
+    public function setConfig(array $config): void;
+
+    /**
+     * @return void
+     */
+    public function start(): void;
+
+    /**
+     * @return void
+     */
+    public function down(): void;
+}

src/Server/TcpServer.php

@@ -0,0 +1,74 @@
+<?php
+
+namespace Asiries335\redisSteamPhp\Server;
+
+use Asiries335\redisSteamPhp\Consumer\ConsumerInterface;
+use React\Socket\ConnectionInterface;
+
+/**
+ *
+ */
+class TcpServer implements ServerInterface
+{
+
+    private array $config;
+
+    /**
+     * @var \React\Socket\ServerInterface
+     */
+    private $tcpServer = null;
+
+
+    /**
+     * @var ConsumerInterface[]
+     */
+    private array $consumers;
+
+    /**
+     * @param array $config
+     * @return void
+     */
+    public function setConfig(array $config): void {
+        $this->config = $config;
+    }
+
+    /**
+     * @return void
+     */
+    public function start(): void {
+        if ($this->tcpServer !== null) {
+            return;
+        }
+
+        $ip = $this->config['ip'] ?? '0.0.0.0';
+        $port = $this->config['port'] ?? '2341';
+
+        $this->tcpServer = new \React\Socket\TcpServer($ip . ':' . $port);
+
+        $this->tcpServer->on('connection', function (ConnectionInterface $connection) {
+
+            $connection->pipe($connection);
+
+            $connection->on('data', function ($payload) {
+                foreach ($this->consumers as $consumer) {
+                    $consumer->handle($payload);
+                }
+            });
+        });
+    }
+
+    /**
+     * @return void
+     */
+    public function down(): void {
+        $this->tcpServer->close();
+    }
+
+    /**
+     * @param ConnectionInterface[] $consumers
+     * @return void
+     */
+    public function setConsumers(array $consumers): void {
+        $this->consumers = $consumers;
+    }
+}

src/ServerProvider.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace Asiries335\redisSteamPhp;
+
+use Asiries335\redisSteamPhp\Consumer\Manager\ConsumerManagerInterface;
+use Asiries335\redisSteamPhp\Server\ServerInterface;
+
+class ServerProvider
+{
+    private ServerInterface $server;
+    private ConsumerManagerInterface $consumerManager;
+
+    /**
+     * @param ServerInterface $server
+     * @param ConsumerManagerInterface $consumerManager
+     *
+     * @return void
+     */
+    public function __construct(ServerInterface $server, ConsumerManagerInterface $consumerManager) {
+        $this->server = $server;
+        $this->consumerManager = $consumerManager;
+    }
+
+    public function boot(): void {
+        $this->server->setConsumers($this->consumerManager->list());
+        $this->server->start();
+    }
+
+
+}