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
|
<?php
namespace React\Http\Io;
use Evenement\EventEmitter;
use React\Stream\ReadableStreamInterface;
use React\Stream\Util;
use React\Stream\WritableStreamInterface;
/**
* [Internal] Limits the amount of data the given stream can emit
*
* This is used internally to limit the size of the underlying connection stream
* to the size defined by the "Content-Length" header of the incoming request.
*
* @internal
*/
class LengthLimitedStream extends EventEmitter implements ReadableStreamInterface
{
private $stream;
private $closed = false;
private $transferredLength = 0;
private $maxLength;
public function __construct(ReadableStreamInterface $stream, $maxLength)
{
$this->stream = $stream;
$this->maxLength = $maxLength;
$this->stream->on('data', array($this, 'handleData'));
$this->stream->on('end', array($this, 'handleEnd'));
$this->stream->on('error', array($this, 'handleError'));
$this->stream->on('close', array($this, 'close'));
}
public function isReadable()
{
return !$this->closed && $this->stream->isReadable();
}
public function pause()
{
$this->stream->pause();
}
public function resume()
{
$this->stream->resume();
}
public function pipe(WritableStreamInterface $dest, array $options = array())
{
Util::pipe($this, $dest, $options);
return $dest;
}
public function close()
{
if ($this->closed) {
return;
}
$this->closed = true;
$this->stream->close();
$this->emit('close');
$this->removeAllListeners();
}
/** @internal */
public function handleData($data)
{
if (($this->transferredLength + \strlen($data)) > $this->maxLength) {
// Only emit data until the value of 'Content-Length' is reached, the rest will be ignored
$data = (string)\substr($data, 0, $this->maxLength - $this->transferredLength);
}
if ($data !== '') {
$this->transferredLength += \strlen($data);
$this->emit('data', array($data));
}
if ($this->transferredLength === $this->maxLength) {
// 'Content-Length' reached, stream will end
$this->emit('end');
$this->close();
$this->stream->removeListener('data', array($this, 'handleData'));
}
}
/** @internal */
public function handleError(\Exception $e)
{
$this->emit('error', array($e));
$this->close();
}
/** @internal */
public function handleEnd()
{
if (!$this->closed) {
$this->handleError(new \Exception('Unexpected end event'));
}
}
}
|
