http/examples/99-server-benchmark-download.php at v1.0.0 · reactphp/http · GitHub

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
126
127
<?php

// $ php examples/99-server-benchmark-download.php 8080
// $ curl http://localhost:8080/10g.bin > /dev/null
// $ wget http://localhost:8080/10g.bin -O /dev/null
// $ ab -n10 -c10 http://localhost:8080/1g.bin
// $ docker run -it --rm --net=host jordi/ab -n100000 -c10 http://localhost:8080/
// $ docker run -it --rm --net=host jordi/ab -n10 -c10 http://localhost:8080/1g.bin

use Evenement\EventEmitter;
use Psr\Http\Message\ServerRequestInterface;
use React\EventLoop\Factory;
use React\Http\Message\Response;
use React\Http\Server;
use React\Stream\ReadableStreamInterface;
use React\Stream\WritableStreamInterface;

require __DIR__ . '/../vendor/autoload.php';

$loop = Factory::create();

/** A readable stream that can emit a lot of data */
class ChunkRepeater extends EventEmitter implements ReadableStreamInterface
{
    private $chunk;
    private $count;
    private $position = 0;
    private $paused = true;
    private $closed = false;

    public function __construct($chunk, $count)
    {
        $this->chunk = $chunk;
        $this->count = $count;
    }

    public function pause()
    {
        $this->paused = true;
    }

    public function resume()
    {
        if (!$this->paused || $this->closed) {
            return;
        }

        // keep emitting until stream is paused
        $this->paused = false;
        while ($this->position < $this->count && !$this->paused) {
            ++$this->position;
            $this->emit('data', array($this->chunk));
        }

        // end once the last chunk has been written
        if ($this->position >= $this->count) {
            $this->emit('end');
            $this->close();
        }
    }

    public function pipe(WritableStreamInterface $dest, array $options = array())
    {
        return;
    }

    public function isReadable()
    {
        return !$this->closed;
    }

    public function close()
    {
        if ($this->closed) {
            return;
        }

        $this->closed = true;
        $this->count = 0;
        $this->paused = true;
        $this->emit('close');
    }

    public function getSize()
    {
        return strlen($this->chunk) * $this->count;
    }
}

$server = new Server($loop, function (ServerRequestInterface $request) use ($loop) {
    switch ($request->getUri()->getPath()) {
        case '/':
            return new Response(
                200,
                array(
                    'Content-Type' => 'text/html'
                ),
                '<html><a href="1g.bin">1g.bin</a><br/><a href="10g.bin">10g.bin</a></html>'
            );
        case '/1g.bin':
            $stream = new ChunkRepeater(str_repeat('.', 1000000), 1000);
            break;
        case '/10g.bin':
            $stream = new ChunkRepeater(str_repeat('.', 1000000), 10000);
            break;
        default:
            return new Response(404);
    }

    $loop->addTimer(0, array($stream, 'resume'));

    return new Response(
        200,
        array(
            'Content-Type' => 'application/octet-data',
            'Content-Length' => $stream->getSize()
        ),
        $stream
    );
});

$socket = new \React\Socket\Server(isset($argv[1]) ? $argv[1] : '0.0.0.0:0', $loop, array('tcp' => array('backlog' => 511)));
$server->listen($socket);

echo 'Listening on ' . str_replace('tcp:', 'http:', $socket->getAddress()) . PHP_EOL;

$loop->run();