dnode-php/src/DNode/DNode.php at d361146e895e51efdb35cfbc42f1c77782988c2d · bergie/dnode-php · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
<?php
namespace DNode;
use Evenement\EventEmitter;
use React\EventLoop\LoopInterface;
use React\Socket\Server;
use React\Socket\ConnectionInterface;

class DNode extends EventEmitter
{
    public $stack = array();

    private $loop;
    private $protocol;

    public function __construct(LoopInterface $loop, $wrapper = null)
    {
        $this->loop = $loop;

        $wrapper = $wrapper ?: new \StdClass();
        $this->protocol = new Protocol($wrapper);
    }

    public function using($middleware)
    {
        $this->stack[] = $middleware;
        return $this;
    }

    public function connect()
    {
        $params = $this->protocol->parseArgs(func_get_args());

        if (!isset($params['scheme'])) {
            $params['scheme'] = 'tcp';
        }

        if (!isset($params['host'])) {
            $params['host'] = '127.0.0.1';
        }

        if (!isset($params['port'])) {
            throw new \Exception("For now we only support connections to a defined port");
        }

        if (!in_array($params['scheme'], stream_get_transports())) {
            $e = new \InvalidArgumentException("Scheme {$params['scheme']} is not supported... are you missing an extension?");

            $this->emit('error', array($e));
        }

        $url = "{$params['scheme']}://{$params['host']}:{$params['port']}";
        $client = @stream_socket_client($url);

        if (!$client) {
            $e = new \RuntimeException("No connection to DNode server in $url");

            $this->emit('error', array($e));

            if (!count($this->listeners('error'))) {
                trigger_error((string) $e, E_USER_ERROR);
            }

            return;
        }

        $conn = new Connection($client, $this->loop);
        $this->handleConnection($conn, $params);
    }

    public function listen()
    {
        $params = $this->protocol->parseArgs(func_get_args());
        if (!isset($params['host'])) {
            $params['host'] = '127.0.0.1';
        }

        if (!isset($params['port'])) {
            throw new \Exception("For now we only support TCP connections to a defined port");
        }

        $that = $this;

        $server = new Server($this->loop);
        $server->on('connection', function ($conn) use ($that, $params) {
            $that->handleConnection($conn, $params);
        });
        $server->listen($params['port'], $params['host']);

        return $server;
    }

    public function handleConnection(ConnectionInterface $conn, $params)
    {
        $client = $this->protocol->create();

        $onReady = isset($params['block']) ? $params['block'] : null;
        $stream = new Stream($this, $client, $onReady);

        $conn->pipe($stream)->pipe($conn);

        $client->start();
    }

    public function end()
    {
        $this->protocol->end();
        $this->emit('end');
    }

    public function close()
    {
        // FIXME: $this->server does not exist
        $this->server->close();
    }
}