Repository navigation
Expand file tree
/
Copy pathSocket.php
More file actions
103 lines (82 loc) · 2.52 KB
/
Copy pathSocket.php
File metadata and controls
103 lines (82 loc) · 2.52 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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
<?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\Socket as CoSocket;
class Socket implements SocketInterface
{
/**
* @var CoSocket
*/
protected $socket;
/**
* @var ProtocolPackerInterface
*/
protected $packer;
/**
* SOCK_STREAM | SOCK_DGRAM.
* @var int
*/
protected $pipeType;
public function __construct(CoSocket $socket, ProtocolPackerInterface $packer, int $pipeType = SOCK_STREAM)
{
$this->socket = $socket;
$this->packer = $packer;
$this->pipeType = $pipeType;
}
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);
}
}