PHP code example of tourze / symfony-workerman-bundle

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

tourze / symfony-workerman-bundle example snippets

namespace App\Worker;

use Tourze\Symfony\WorkermanBundle\Contracts\ConnectableInterface;
use Tourze\Symfony\WorkermanBundle\Contracts\WorkerBuilderInterface;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;

class EchoWorker implements WorkerBuilderInterface, ConnectableInterface
{
    public function getName(): string
    {
        return 'echo-server';
    }

    public function getTransport(): string
    {
        return 'tcp';
    }

    public function getListenIp(): string
    {
        return '0.0.0.0';
    }

    public function getListenPort(): int
    {
        return 2345;
    }

    public function onWorkerStart(Worker $worker): void
    {
        echo "Worker started\n";
    }

    public function onWorkerStop(Worker $worker): void
    {
        echo "Worker stopped\n";
    }

    public function onWorkerReload(Worker $worker): void
    {
        echo "Worker reloaded\n";
    }

    public function onConnect(TcpConnection $connection): void
    {
        echo "New connection from {$connection->getRemoteIp()}\n";
    }

    public function onMessage(TcpConnection $connection, mixed $data): void
    {
        $connection->send("Echo: {$data}");
    }

    public function onClose(TcpConnection $connection): void
    {
        echo "Connection closed\n";
    }

    public function onError(TcpConnection $connection, int $code, string $msg): void
    {
        echo "Error: {$msg} (code: {$code})\n";
    }
}

namespace App\Timer;

use Tourze\Symfony\WorkermanBundle\Contracts\TimerInterface;

class HeartbeatTimer implements TimerInterface
{
    public function getExpression(): string
    {
        return '*/30 * * * * *'; // Every 30 seconds
    }

    public function execute(): void
    {
        echo "Heartbeat at " . date('Y-m-d H:i:s') . "\n";
    }
}

namespace App\EventListener;

use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Tourze\Symfony\WorkermanBundle\Event\TcpWorkerConnectEvent;
use Tourze\Symfony\WorkermanBundle\Event\TcpWorkerMessageEvent;
use Tourze\Symfony\WorkermanBundle\Event\TcpWorkerCloseEvent;

class TcpServerListener implements EventSubscriberInterface
{
    public static function getSubscribedEvents(): array
    {
        return [
            TcpWorkerConnectEvent::class => 'onConnect',
            TcpWorkerMessageEvent::class => 'onMessage',
            TcpWorkerCloseEvent::class => 'onClose',
        ];
    }
    
    public function onConnect(TcpWorkerConnectEvent $event): void
    {
        $connection = $event->getConnection();
        $connection->send("Welcome to TCP Server!\n");
        
        // Store connection metadata
        $connection->uid = uniqid();
        echo "Client {$connection->uid} connected from {$connection->getRemoteIp()}\n";
    }
    
    public function onMessage(TcpWorkerMessageEvent $event): void
    {
        $connection = $event->getConnection();
        $message = $event->getMessage();
        
        // Echo the message back
        $connection->send("Echo: {$message}");
        
        // Broadcast to all connections
        foreach ($connection->worker->connections as $conn) {
            if ($conn->id !== $connection->id) {
                $conn->send("Client {$connection->uid} says: {$message}");
            }
        }
    }
    
    public function onClose(TcpWorkerCloseEvent $event): void
    {
        $connection = $event->getConnection();
        echo "Client {$connection->uid} disconnected\n";
    }
}

namespace App\Protocol;

use Workerman\Protocols\ProtocolInterface;

class CustomProtocol implements ProtocolInterface
{
    public static function input($recv_buffer, $connection)
    {
        // Return 0 if not enough data for complete package
        // Return package length if complete package received
        $recv_len = strlen($recv_buffer);
        if ($recv_len < 4) {
            return 0;
        }
        
        $unpack_data = unpack('Nlength', $recv_buffer);
        $package_length = $unpack_data['length'];
        
        if ($recv_len < $package_length + 4) {
            return 0;
        }
        
        return $package_length + 4;
    }
    
    public static function decode($recv_buffer, $connection)
    {
        $data = substr($recv_buffer, 4);
        return json_decode($data, true);
    }
    
    public static function encode($data, $connection)
    {
        $json_data = json_encode($data);
        return pack('N', strlen($json_data)) . $json_data;
    }
}

namespace App\Worker;

use Tourze\Symfony\WorkermanBundle\Contracts\WorkerBuilderInterface;
use Workerman\Worker;

class AdvancedWorker implements WorkerBuilderInterface
{
    public function getName(): string
    {
        return 'advanced-worker';
    }
    
    public function onWorkerStart(Worker $worker): void
    {
        // Set custom properties
        $worker->reusePort = true;
        $worker->count = 4; // Override default CPU count
        
        // Initialize resources
        $this->initializeResources();
    }
    
    public function onWorkerStop(Worker $worker): void
    {
        $this->cleanup();
    }
    
    public function onWorkerReload(Worker $worker): void
    {
        $this->reloadConfiguration();
    }
    
    private function initializeResources(): void
    {
        // Initialize databases, caches, etc.
    }
    
    private function cleanup(): void
    {
        // Cleanup resources
    }
    
    private function reloadConfiguration(): void
    {
        // Reload config without restart
    }
}

bash
php bin/console workerman:run <action> [options]

bash
php bin/console workerman:tcp [options]