-
Notifications
You must be signed in to change notification settings - Fork 52
/
Socket.php
125 lines (101 loc) · 2.72 KB
/
Socket.php
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
<?php
namespace Cassandra\Connection;
class Socket {
/**
* @var resource
*/
protected $_socket;
/**
* @var array
*/
protected $_options = [
'host' => null,
'port' => 9042,
'username' => null,
'password' => null,
'socket' => [
SO_RCVTIMEO => ["sec" => 30, "usec" => 0],
SO_SNDTIMEO => ["sec" => 5, "usec" => 0],
],
];
/**
* @param array $options
*/
public function __construct(array $options) {
if (isset($options['socket'])) {
$options['socket'] += $this->_options['socket'];
}
$this->_options = array_merge($this->_options, $options);
$this->_connect();
}
/**
*
* @throws SocketException
* @return resource
*/
protected function _connect() {
if (!empty($this->_socket)) return $this->_socket;
$this->_socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
if ($this->_socket === false){
$errorCode = socket_last_error($this->_socket);
throw new SocketException(socket_strerror($errorCode), $errorCode);
}
socket_set_option($this->_socket, SOL_TCP, TCP_NODELAY, 1);
foreach($this->_options['socket'] as $optname => $optval)
socket_set_option($this->_socket, SOL_SOCKET, $optname, $optval);
$result = socket_connect($this->_socket, $this->_options['host'], $this->_options['port']);
if ($result === false){
$errorCode = socket_last_error($this->_socket);
//Unable to connect to Cassandra node: {$this->_options['host']}:{$this->_options['port']}
throw new SocketException(socket_strerror($errorCode), $errorCode);
}
}
/**
* @return array
*/
public function getOptions() {
return $this->_options;
}
/**
* @param $length
* @throws SocketException
* @return string
*/
public function read($length) {
$data = socket_read($this->_socket, $length);
if ($data === false){
$errorCode = socket_last_error($this->_socket);
throw new SocketException(socket_strerror($errorCode), $errorCode);
}
$remainder = $length - strlen($data);
while($remainder > 0) {
$readData = socket_read($this->_socket, $remainder);
if ($readData === false){
$errorCode = socket_last_error($this->_socket);
throw new SocketException(socket_strerror($errorCode), $errorCode);
}
$data .= $readData;
$remainder -= strlen($readData);
}
return $data;
}
/**
*
* @param string $binary
* @throws SocketException
*/
public function write($binary){
do{
$sentBytes = socket_write($this->_socket, $binary);
if ($sentBytes === false){
$errorCode = socket_last_error($this->_socket);
throw new SocketException(socket_strerror($errorCode), $errorCode);
}
$binary = substr($binary, $sentBytes);
}
while(!empty($binary));
}
public function close(){
socket_shutdown($this->_socket);
}
}