Skip to content

Commit 6d6a843

Browse files
cbodencboden
authored
Merge pull request #52 from clue-labs/write-feof
Only emit permanent errors and otherwise keep retrying
2 parents 7782ea0 + d9e7f89 commit 6d6a843

3 files changed

Lines changed: 167 additions & 2 deletions

File tree

‎src/Buffer.php‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,14 @@ public function handleWrite()
8787

8888
restore_error_handler();
8989

90-
if ($this->lastError['number'] > 0) {
90+
// Only report errors if *nothing* could be sent.
91+
// Any hard (permanent) error will fail to send any data at all.
92+
// Sending excessive amounts of data will only flush *some* data and then
93+
// report a temporary error (EAGAIN) which we do not raise here in order
94+
// to keep the stream open for further tries to write.
95+
// Should this turn out to be a permanent error later, it will eventually
96+
// send *nothing* and we can detect this.
97+
if ($sent === 0 && $this->lastError['number'] > 0) {
9198
$this->emit('error', array(
9299
new \ErrorException(
93100
$this->lastError['message'],
@@ -102,7 +109,7 @@ public function handleWrite()
102109
return;
103110
}
104111

105-
if ($sent === false) {
112+
if ($sent === 0) {
106113
$this->emit('error', array(new \RuntimeException('Send failed'), $this));
107114
return;
108115
}

‎tests/BufferTest.php‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,28 @@ public function testWriteReturnsFalseWhenBufferIsFull()
8484
$this->assertFalse($buffer->write("bar\n"));
8585
}
8686

87+
/**
88+
* @covers React\Stream\Buffer::write
89+
* @covers React\Stream\Buffer::handleWrite
90+
*/
91+
public function testWriteEmitsErrorWhenResourceIsNotWritable()
92+
{
93+
if (defined('HHVM_VERSION')) {
94+
// via https://github.com/reactphp/stream/pull/52/files#r75493076
95+
$this->markTestSkipped('HHVM allows writing to read-only memory streams');
96+
}
97+
98+
$stream = fopen('php://temp', 'r');
99+
$loop = $this->createLoopMock();
100+
101+
$buffer = new Buffer($stream, $loop);
102+
$buffer->on('error', $this->expectCallableOnce());
103+
//$buffer->on('close', $this->expectCallableOnce());
104+
105+
$buffer->write('hello');
106+
$buffer->handleWrite();
107+
}
108+
87109
/**
88110
* @covers React\Stream\Buffer::write
89111
* @covers React\Stream\Buffer::handleWrite

‎tests/StreamIntegrationTest.php‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,46 @@ public function testBufferReadsLargeChunks($condition, $loopFactory)
5656
$this->assertEquals($testString, $buffer);
5757
}
5858

59+
/**
60+
* @dataProvider loopProvider
61+
*/
62+
public function testWriteLargeChunk($condition, $loopFactory)
63+
{
64+
if (true !== $condition()) {
65+
return $this->markTestSkipped('Loop implementation not available');
66+
}
67+
68+
$loop = $loopFactory();
69+
70+
list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
71+
72+
$streamA = new Stream($sockA, $loop);
73+
$streamB = new Stream($sockB, $loop);
74+
75+
// limit seems to be 192 KiB
76+
$size = 256 * 1024;
77+
78+
// sending side sends and expects clean close with no errors
79+
$streamA->end(str_repeat('*', $size));
80+
$streamA->on('close', $this->expectCallableOnce());
81+
$streamA->on('error', $this->expectCallableNever());
82+
83+
// receiving side counts bytes and expects clean close with no errors
84+
$received = 0;
85+
$streamB->on('data', function ($chunk) use (&$received) {
86+
$received += strlen($chunk);
87+
});
88+
$streamB->on('close', $this->expectCallableOnce());
89+
$streamB->on('error', $this->expectCallableNever());
90+
91+
$loop->run();
92+
93+
$streamA->close();
94+
$streamB->close();
95+
96+
$this->assertEquals($size, $received);
97+
}
98+
5999
/**
60100
* @dataProvider loopProvider
61101
*/
@@ -83,4 +123,100 @@ public function testDoesNotEmitDataIfNothingHasBeenWritten($condition, $loopFact
83123
$streamA->close();
84124
$streamB->close();
85125
}
126+
127+
/**
128+
* @dataProvider loopProvider
129+
*/
130+
public function testDoesNotWriteDataIfRemoteSideFromPairHasBeenClosed($condition, $loopFactory)
131+
{
132+
if (true !== $condition()) {
133+
return $this->markTestSkipped('Loop implementation not available');
134+
}
135+
136+
$loop = $loopFactory();
137+
138+
list($sockA, $sockB) = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
139+
140+
$streamA = new Stream($sockA, $loop);
141+
$streamB = new Stream($sockB, $loop);
142+
143+
// end streamA without writing any data
144+
$streamA->pause();
145+
$streamA->write('hello');
146+
$streamA->on('close', $this->expectCallableOnce());
147+
148+
$streamB->on('data', $this->expectCallableNever());
149+
$streamB->close();
150+
151+
$loop->run();
152+
153+
$streamA->close();
154+
$streamB->close();
155+
}
156+
157+
/**
158+
* @dataProvider loopProvider
159+
*/
160+
public function testDoesNotWriteDataIfServerSideHasBeenClosed($condition, $loopFactory)
161+
{
162+
if (true !== $condition()) {
163+
return $this->markTestSkipped('Loop implementation not available');
164+
}
165+
166+
$loop = $loopFactory();
167+
168+
$server = stream_socket_server('tcp://localhost:0');
169+
170+
$client = stream_socket_client(stream_socket_get_name($server, false));
171+
$peer = stream_socket_accept($server);
172+
173+
$streamA = new Stream($client, $loop);
174+
$streamB = new Stream($peer, $loop);
175+
176+
// end streamA without writing any data
177+
$streamA->pause();
178+
$streamA->write('hello');
179+
$streamA->on('close', $this->expectCallableOnce());
180+
181+
$streamB->on('data', $this->expectCallableNever());
182+
$streamB->close();
183+
184+
$loop->run();
185+
186+
$streamA->close();
187+
$streamB->close();
188+
}
189+
190+
/**
191+
* @dataProvider loopProvider
192+
*/
193+
public function testDoesNotWriteDataIfClientSideHasBeenClosed($condition, $loopFactory)
194+
{
195+
if (true !== $condition()) {
196+
return $this->markTestSkipped('Loop implementation not available');
197+
}
198+
199+
$loop = $loopFactory();
200+
201+
$server = stream_socket_server('tcp://localhost:0');
202+
203+
$client = stream_socket_client(stream_socket_get_name($server, false));
204+
$peer = stream_socket_accept($server);
205+
206+
$streamA = new Stream($peer, $loop);
207+
$streamB = new Stream($client, $loop);
208+
209+
// end streamA without writing any data
210+
$streamA->pause();
211+
$streamA->write('hello');
212+
$streamA->on('close', $this->expectCallableOnce());
213+
214+
$streamB->on('data', $this->expectCallableNever());
215+
$streamB->close();
216+
217+
$loop->run();
218+
219+
$streamA->close();
220+
$streamB->close();
221+
}
86222
}

0 commit comments

Comments
 (0)