[deprecated] moved to https://github.com/clue/datagram [DRAFT] UDP DatagramClient and DatagramServer built on top of reactphp

DatagramClient.php

<?php

class DatagramClient extends DatagramSocketReadable
{   
    public function __construct($loop, $socket, $address = null)
    {
        if ($address === null) {
            $address = stream_socket_get_name($socket, true);
        }
        parent::__construct($loop, $socket, $address);
        $this->resume();
    }
}

DatagramFactory.php

<?php

class DatagramFactory
{
    public function createClient(LoopInterface $loop, Resolver $resolver, $host, $port)
    {
        $factory = $this;
        return $this->resolve($resolver, $host)->then(function ($ip) use ($loop, $port, $factory) {
            $address = $factory->createAddress($ip, $port);
            $socket = stream_socket_client('udp://' . $address, $errno, $errstr)

            return new DatagramClient($loop, $socket, $address);
        });
    }

    public function createServer(LoopInterface $loop, $port, $host = '127.0.0.1')
    {
        $address = $this->createAddress($host, $port);
        
        $socket = stream_socket_server("udp://" . $address, $errno, $errstr, STREAM_SERVER_BIND);
        if (!$socket) {
            die("$errstr ($errno)");
        }
        return When::resolve(new DatagramServer($loop, $socket, $address));
    }
    
    protected function resolve($resolver, $host)
    {
        // todo: if there's no need to resolve the address:
        if (false) {
            return When::resolve($host);
        }
        return $resolver->resolve($host);
    }
    
    public function createAddress($host, $port)
    {
        $address = $host;
        if (strpos($host, ':') !== false) {
            // enclose IPv6 address in square brackets
            $address = '[' . $host . ']';
        }
        $address .= $port;
        return $address;
    }
}

DatagramServer.php

<?php

// TODO: should not provide a send() method
class DatagramServer extends DatagramSocketReadable
{
    public function __construct(LoopInterface $loop, $socket, $address = null)
    {
        if ($address === null) {
            $address = stream_socket_get_name($socket, false);
        }
        parent::__construct($loop, $socket, $address);
        $this->resume();
    }
}

DatagramSocket.php

<?php

class DatagramSocket extends EventEmitter
{
    protected $loop;
    protected $socket;
    protected $address;
    
    public function __construct($loop, $socket, $address)
    {
        $this->loop = $loop;
        $this->socket = $socket;
        $this->address = $address;
    }
    
    public function getAddress()
    {
        // TODO: doc comment suggests IPv6 address is not enclosed in square brackets?
        return $this->address;
    }
    
    public function getPort()
    {
        return (int)substr($this->address, strrpos($this->address, ':') + 1);
    }
    
    public function getHost(){
        return trim(substr($this->address, 0, strrpos($this->address, ':'), '[]');
    }
    
    public function send($data)
    {
        stream_socket_sendto($this->socket, $data, 0, $this->address);
    }
}

DatagramSocketReadable.php

<?php

/** @event message */
// interface similar to Stream
// TODO: implment EventEmitter only here?
class DatagramSocketReadable extends DatagramSocket
{
    public $bufferSize = 1500;
    
    public function pause()
    {
        $this->loop->removeReadStream($this->socket);
    }
    
    public function resume()
    {
        $this->loop->addReadStream($this->socket, array($this, 'onReceive'));
    }
    
    public function onReceive($message)
    {
        $data = stream_socket_recvfrom($this->socket, $this->bufferSize, 0, $peer);
        
        // create remote socket that does NOT have a dedicated readable method (see thie main DatagramSocket instead)
        $remote = new DatagramSocket($this->socket, $peer);
            
        $this->emit('message', array($data, $socket);
    }
}

example-client.php

<?php

$loop = React\Loop\Factory::create();

$factory = Resolver\Factory();
$resolver = $factory->createCached($loop, '8.8.8.8');

$factory = new DatagramFactory();

$factory->createClient($resolver, 'localhost', 1234)->then(function ($client) use ($loop) {
    $client->send('first');

    $client->on('message', function($message, $server) {
        //$remote->send() is same as $client->send()
        
        echo 'received "' . $message . '" from ' . $server->getAddress() . PHP_EOL;
    });
    
    $n = 0;
    $loop->addPeriodicTimer(2.0, function() use ($client, &$n) {
        $client->send('tick' . $n);
    });
});

$loop->run();

example-server.php

<?php

$loop = React\Loop\Factory::create();

$factory = new DatagramFactory();

$factory->createServer($loop, 1234)->then(function ($server) {
    $server->on('message', function($data, $client) {
        $client->send('hello '.$client->getAddress().'! echo: '.$message);
        
        // $server->send() is not available here
    });
});

$loop->run();