stream/src/CompositeStream.php at 5047943c375e568f52ae67bb3a5abfb60af65025 · clue-labs/stream · 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
<?php

namespace React\Stream;

use Evenement\EventEmitter;

final class CompositeStream extends EventEmitter implements DuplexStreamInterface
{
    /** @var ReadableStreamInterface */
    private $readable;

    /** @var WritableStreamInterface */
    private $writable;

    /** @var bool */
    private $closed = false;

    public function __construct(ReadableStreamInterface $readable, WritableStreamInterface $writable)
    {
        $this->readable = $readable;
        $this->writable = $writable;

        if (!$readable->isReadable() || !$writable->isWritable()) {
            $this->close();
            return;
        }

        Util::forwardEvents($this->readable, $this, ['data', 'end', 'error']);
        Util::forwardEvents($this->writable, $this, ['drain', 'error', 'pipe']);

        $this->readable->on('close', [$this, 'close']);
        $this->writable->on('close', [$this, 'close']);
    }

    public function isReadable(): bool
    {
        return $this->readable->isReadable();
    }

    public function pause(): void
    {
        $this->readable->pause();
    }

    public function resume(): void
    {
        if (!$this->writable->isWritable()) {
            return;
        }

        $this->readable->resume();
    }

    public function pipe(WritableStreamInterface $dest, array $options = []): WritableStreamInterface
    {
        return Util::pipe($this, $dest, $options);
    }

    public function isWritable(): bool
    {
        return $this->writable->isWritable();
    }

    public function write($data): bool
    {
        return $this->writable->write($data);
    }

    public function end($data = null): void
    {
        $this->readable->pause();
        $this->writable->end($data);
    }

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

        $this->closed = true;
        $this->readable->close();
        $this->writable->close();

        $this->emit('close');
        $this->removeAllListeners();
    }
}