Skip to content

Commit af0cce3

Browse files
clueclue
committed
Improve socket error codes by preserving upstreaming errnos
1 parent e57ccc2 commit af0cce3

4 files changed

Lines changed: 173 additions & 13 deletions

File tree

‎src/Factory.php‎

Lines changed: 21 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -111,9 +111,16 @@ function () use ($client) {
111111
function (\Exception $e) use ($client, $uri) {
112112
$client->close();
113113

114+
$const = '';
115+
$errno = $e->getCode();
116+
if ($errno === 0) {
117+
$const = ' (EACCES)';
118+
$errno = $e->getCode() ?: (defined('SOCKET_EACCES') ? SOCKET_EACCES : 13);
119+
}
120+
114121
throw new \RuntimeException(
115-
'Connection to ' . $uri . ' failed during AUTH command: ' . $e->getMessage() . ' (EACCES)',
116-
defined('SOCKET_EACCES') ? SOCKET_EACCES : 13,
122+
'Connection to ' . $uri . ' failed during AUTH command: ' . $e->getMessage() . $const,
123+
$errno,
117124
$e
118125
);
119126
}
@@ -132,9 +139,19 @@ function () use ($client) {
132139
function (\Exception $e) use ($client, $uri) {
133140
$client->close();
134141

142+
$const = '';
143+
$errno = $e->getCode();
144+
if ($errno === 0 && strpos($e->getMessage(), 'NOAUTH ') === 0) {
145+
$const = ' (EACCES)';
146+
$errno = defined('SOCKET_EACCES') ? SOCKET_EACCES : 13;
147+
} elseif ($errno === 0) {
148+
$const = ' (ENOENT)';
149+
$errno = defined('SOCKET_ENOENT') ? SOCKET_ENOENT : 2;
150+
}
151+
135152
throw new \RuntimeException(
136-
'Connection to ' . $uri . ' failed during SELECT command: ' . $e->getMessage() . ' (ENOENT)',
137-
defined('SOCKET_ENOENT') ? SOCKET_ENOENT : 2,
153+
'Connection to ' . $uri . ' failed during SELECT command: ' . $e->getMessage() . $const,
154+
$errno,
138155
$e
139156
);
140157
}

‎src/StreamingClient.php‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ public function handleMessage(ModelInterface $message)
146146
}
147147

148148
$request = array_shift($this->requests);
149-
/* @var $request Deferred */
149+
assert($request instanceof Deferred);
150150

151151
if ($message instanceof ErrorReply) {
152152
$request->reject($message);
@@ -177,18 +177,27 @@ public function close()
177177
$this->ending = true;
178178
$this->closed = true;
179179

180+
$remoteClosed = $this->stream->isReadable() === false && $this->stream->isWritable() === false;
180181
$this->stream->close();
181182

182183
$this->emit('close');
183184

184185
// reject all remaining requests in the queue
185-
while($this->requests) {
186+
while ($this->requests) {
186187
$request = array_shift($this->requests);
187-
/* @var $request Deferred */
188-
$request->reject(new \RuntimeException(
189-
'Connection closing (ENOTCONN)',
190-
defined('SOCKET_ENOTCONN') ? SOCKET_ENOTCONN : 107
191-
));
188+
assert($request instanceof Deferred);
189+
190+
if ($remoteClosed) {
191+
$request->reject(new \RuntimeException(
192+
'Connection closed by peer (ECONNRESET)',
193+
defined('SOCKET_ECONNRESET') ? SOCKET_ECONNRESET : 104
194+
));
195+
} else {
196+
$request->reject(new \RuntimeException(
197+
'Connection closing (ECONNABORTED)',
198+
defined('SOCKET_ECONNABORTED') ? SOCKET_ECONNABORTED : 103
199+
));
200+
}
192201
}
193202
}
194203
}

‎tests/FactoryStreamingClientTest.php‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,44 @@ public function testWillRejectAndCloseAutomaticallyWhenAuthCommandReceivesErrorR
212212
));
213213
}
214214

215+
public function testWillRejectAndCloseAutomaticallyWhenConnectionIsClosedWhileWaitingForAuthCommand()
216+
{
217+
$closeHandler = null;
218+
$stream = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
219+
$stream->expects($this->once())->method('write')->with("*2\r\n$4\r\nauth\r\n$5\r\nworld\r\n");
220+
$stream->expects($this->once())->method('close');
221+
$stream->expects($this->exactly(2))->method('on')->withConsecutive(
222+
array('data', $this->anything()),
223+
array('close', $this->callback(function ($arg) use (&$closeHandler) {
224+
$closeHandler = $arg;
225+
return true;
226+
}))
227+
);
228+
229+
$this->connector->expects($this->once())->method('connect')->willReturn(Promise\resolve($stream));
230+
$promise = $this->factory->createClient('redis://:world@localhost');
231+
232+
$this->assertTrue(is_callable($closeHandler));
233+
$stream->expects($this->once())->method('isReadable')->willReturn(false);
234+
$stream->expects($this->once())->method('isWritable')->willReturn(false);
235+
call_user_func($closeHandler);
236+
237+
$promise->then(null, $this->expectCallableOnceWith(
238+
$this->logicalAnd(
239+
$this->isInstanceOf('RuntimeException'),
240+
$this->callback(function (\Exception $e) {
241+
return $e->getMessage() === 'Connection to redis://:***@localhost failed during AUTH command: Connection closed by peer (ECONNRESET)';
242+
}),
243+
$this->callback(function (\Exception $e) {
244+
return $e->getCode() === (defined('SOCKET_ECONNRESET') ? SOCKET_ECONNRESET : 104);
245+
}),
246+
$this->callback(function (\Exception $e) {
247+
return $e->getPrevious()->getMessage() === 'Connection closed by peer (ECONNRESET)';
248+
})
249+
)
250+
));
251+
}
252+
215253
public function testWillWriteSelectCommandIfRedisUnixUriContainsDbQueryParameter()
216254
{
217255
$stream = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
@@ -279,6 +317,80 @@ public function testWillRejectAndCloseAutomaticallyWhenSelectCommandReceivesErro
279317
));
280318
}
281319

320+
public function testWillRejectAndCloseAutomaticallyWhenSelectCommandReceivesAuthErrorResponseIfRedisUriContainsPath()
321+
{
322+
$dataHandler = null;
323+
$stream = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
324+
$stream->expects($this->once())->method('write')->with("*2\r\n$6\r\nselect\r\n$3\r\n123\r\n");
325+
$stream->expects($this->once())->method('close');
326+
$stream->expects($this->exactly(2))->method('on')->withConsecutive(
327+
array('data', $this->callback(function ($arg) use (&$dataHandler) {
328+
$dataHandler = $arg;
329+
return true;
330+
})),
331+
array('close', $this->anything())
332+
);
333+
334+
$this->connector->expects($this->once())->method('connect')->willReturn(Promise\resolve($stream));
335+
$promise = $this->factory->createClient('redis://localhost/123');
336+
337+
$this->assertTrue(is_callable($dataHandler));
338+
$dataHandler("-NOAUTH Authentication required.\r\n");
339+
340+
$promise->then(null, $this->expectCallableOnceWith(
341+
$this->logicalAnd(
342+
$this->isInstanceOf('RuntimeException'),
343+
$this->callback(function (\Exception $e) {
344+
return $e->getMessage() === 'Connection to redis://localhost/123 failed during SELECT command: NOAUTH Authentication required. (EACCES)';
345+
}),
346+
$this->callback(function (\Exception $e) {
347+
return $e->getCode() === (defined('SOCKET_EACCES') ? SOCKET_EACCES : 13);
348+
}),
349+
$this->callback(function (\Exception $e) {
350+
return $e->getPrevious()->getMessage() === 'NOAUTH Authentication required.';
351+
})
352+
)
353+
));
354+
}
355+
356+
public function testWillRejectAndCloseAutomaticallyWhenConnectionIsClosedWhileWaitingForSelectCommand()
357+
{
358+
$closeHandler = null;
359+
$stream = $this->getMockBuilder('React\Socket\ConnectionInterface')->getMock();
360+
$stream->expects($this->once())->method('write')->with("*2\r\n$6\r\nselect\r\n$3\r\n123\r\n");
361+
$stream->expects($this->once())->method('close');
362+
$stream->expects($this->exactly(2))->method('on')->withConsecutive(
363+
array('data', $this->anything()),
364+
array('close', $this->callback(function ($arg) use (&$closeHandler) {
365+
$closeHandler = $arg;
366+
return true;
367+
}))
368+
);
369+
370+
$this->connector->expects($this->once())->method('connect')->willReturn(Promise\resolve($stream));
371+
$promise = $this->factory->createClient('redis://localhost/123');
372+
373+
$this->assertTrue(is_callable($closeHandler));
374+
$stream->expects($this->once())->method('isReadable')->willReturn(false);
375+
$stream->expects($this->once())->method('isWritable')->willReturn(false);
376+
call_user_func($closeHandler);
377+
378+
$promise->then(null, $this->expectCallableOnceWith(
379+
$this->logicalAnd(
380+
$this->isInstanceOf('RuntimeException'),
381+
$this->callback(function (\Exception $e) {
382+
return $e->getMessage() === 'Connection to redis://localhost/123 failed during SELECT command: Connection closed by peer (ECONNRESET)';
383+
}),
384+
$this->callback(function (\Exception $e) {
385+
return $e->getCode() === (defined('SOCKET_ECONNRESET') ? SOCKET_ECONNRESET : 104);
386+
}),
387+
$this->callback(function (\Exception $e) {
388+
return $e->getPrevious()->getMessage() === 'Connection closed by peer (ECONNRESET)';
389+
})
390+
)
391+
));
392+
}
393+
282394
public function testWillRejectIfConnectorRejects()
283395
{
284396
$this->connector->expects($this->once())->method('connect')->with('127.0.0.1:2')->willReturn(Promise\reject(new \RuntimeException('Foo', 42)));

‎tests/StreamingClientTest.php‎

Lines changed: 24 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -178,10 +178,32 @@ public function testClosingClientRejectsAllRemainingRequests()
178178
$this->logicalAnd(
179179
$this->isInstanceOf('RuntimeException'),
180180
$this->callback(function (\RuntimeException $e) {
181-
return $e->getMessage() === 'Connection closing (ENOTCONN)';
181+
return $e->getMessage() === 'Connection closing (ECONNABORTED)';
182182
}),
183183
$this->callback(function (\RuntimeException $e) {
184-
return $e->getCode() === (defined('SOCKET_ENOTCONN') ? SOCKET_ENOTCONN : 107);
184+
return $e->getCode() === (defined('SOCKET_ECONNABORTED') ? SOCKET_ECONNABORTED : 103);
185+
})
186+
)
187+
));
188+
}
189+
190+
public function testClosingStreamRejectsAllRemainingRequests()
191+
{
192+
$this->stream = new ThroughStream();
193+
$this->parser->expects($this->once())->method('pushIncoming')->willReturn(array());
194+
$this->client = new StreamingClient($this->stream, $this->parser, $this->serializer);
195+
196+
$promise = $this->client->ping();
197+
$this->stream->close();
198+
199+
$promise->then(null, $this->expectCallableOnceWith(
200+
$this->logicalAnd(
201+
$this->isInstanceOf('RuntimeException'),
202+
$this->callback(function (\RuntimeException $e) {
203+
return $e->getMessage() === 'Connection closed by peer (ECONNRESET)';
204+
}),
205+
$this->callback(function (\RuntimeException $e) {
206+
return $e->getCode() === (defined('SOCKET_ECONNRESET') ? SOCKET_ECONNRESET : 104);
185207
})
186208
)
187209
));

0 commit comments

Comments
 (0)