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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
|
<?php
namespace React\Socket;
use Evenement\EventEmitter;
use React\EventLoop\LoopInterface;
use InvalidArgumentException;
use RuntimeException;
/**
* The `UnixServer` class implements the `ServerInterface` and
* is responsible for accepting plaintext connections on unix domain sockets.
*
* ```php
* $server = new UnixServer('unix:///tmp/app.sock', $loop);
* ```
*
* See also the `ServerInterface` for more details.
*
* @see ServerInterface
* @see ConnectionInterface
*/
final class UnixServer extends EventEmitter implements ServerInterface
{
private $master;
private $loop;
private $listening = false;
/**
* Creates a plaintext socket server and starts listening on the given unix socket
*
* This starts accepting new incoming connections on the given address.
* See also the `connection event` documented in the `ServerInterface`
* for more details.
*
* ```php
* $server = new UnixServer('unix:///tmp/app.sock', $loop);
* ```
*
* @param string $path
* @param LoopInterface $loop
* @param array $context
* @throws InvalidArgumentException if the listening address is invalid
* @throws RuntimeException if listening on this address fails (already in use etc.)
*/
public function __construct($path, LoopInterface $loop, array $context = array())
{
$this->loop = $loop;
if (strpos($path, '://') === false) {
$path = 'unix://' . $path;
} elseif (substr($path, 0, 7) !== 'unix://') {
throw new InvalidArgumentException('Given URI "' . $path . '" is invalid');
}
$this->master = @stream_socket_server(
$path,
$errno,
$errstr,
STREAM_SERVER_BIND | STREAM_SERVER_LISTEN,
stream_context_create(array('socket' => $context))
);
if (false === $this->master) {
throw new RuntimeException('Failed to listen on unix domain socket "' . $path . '": ' . $errstr, $errno);
}
stream_set_blocking($this->master, 0);
$this->resume();
}
public function getAddress()
{
if (!is_resource($this->master)) {
return null;
}
return 'unix://' . stream_socket_get_name($this->master, false);
}
public function pause()
{
if (!$this->listening) {
return;
}
$this->loop->removeReadStream($this->master);
$this->listening = false;
}
public function resume()
{
if ($this->listening || !is_resource($this->master)) {
return;
}
$that = $this;
$this->loop->addReadStream($this->master, function ($master) use ($that) {
$newSocket = @stream_socket_accept($master);
if (false === $newSocket) {
$that->emit('error', array(new RuntimeException('Error accepting new connection')));
return;
}
$that->handleConnection($newSocket);
});
$this->listening = true;
}
public function close()
{
if (!is_resource($this->master)) {
return;
}
$this->pause();
fclose($this->master);
$this->removeAllListeners();
}
/** @internal */
public function handleConnection($socket)
{
$connection = new Connection($socket, $this->loop);
$connection->unix = true;
$this->emit('connection', array(
$connection
));
}
}
|
