Repository navigation
Expand file tree
/
Copy pathSocket.php
More file actions
85 lines (66 loc) · 2.22 KB
/
Copy pathSocket.php
File metadata and controls
85 lines (66 loc) · 2.22 KB
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
<?php
declare(strict_types=1);
/**
* This file is part of Hyperf.
*
* @link https://www.hyperf.io
* @document https://hyperf.wiki
* @contact [email protected]
* @license https://github.com/hyperf/hyperf/blob/master/LICENSE
*/
namespace Hyperf\Socket;
use Hyperf\Protocol\ProtocolPackerInterface;
use Hyperf\Socket\Exception\SocketException;
use Swoole\Coroutine;
class Socket implements SocketInterface
{
public function __construct(protected Coroutine\Socket $socket, protected ProtocolPackerInterface $packer, protected int $pipeType = SOCK_STREAM)
{
}
public function send($data, float $timeout = -1)
{
$string = $this->packer->pack($data);
$length = $this->socket->sendAll($string, $timeout);
if ($length !== strlen($string)) {
throw new SocketException('Send failed: ' . $this->socket->errMsg);
}
return $length;
}
public function recv(float $timeout = -1)
{
if ($this->pipeType === SOCK_DGRAM) {
return $this->recvDgram($timeout);
}
return $this->recvStream($timeout);
}
protected function recvStream(float $timeout)
{
$head = $this->socket->recvAll($this->packer::HEAD_LENGTH, $timeout);
if ($head === false) {
return false;
}
if (strlen($head) !== $this->packer::HEAD_LENGTH) {
throw new SocketException('Receive head failed: ' . $this->socket->errMsg);
}
$length = $this->packer->length($head);
if ($length === 0) {
throw new SocketException('Recv body failed: body length is zero.');
}
$body = $this->socket->recvAll($length, $timeout);
if ($length !== strlen($body)) {
throw new SocketException('Receive body failed: ' . $this->socket->errMsg);
}
return $this->packer->unpack($head . $body);
}
protected function recvDgram(float $timeout)
{
$body = $this->socket->recv(65536, $timeout);
if ($body === false) {
return false;
}
if (strlen($body) === 0) {
throw new SocketException('Receive body failed: body length is zero.');
}
return $this->packer->unpack($body);
}
}