input = $input; $this->input->on('data', array($this, 'handleData')); $this->input->on('end', array($this, 'handleEnd')); $this->input->on('error', array($this, 'handleError')); $this->input->on('close', array($this, 'close')); } public function isReadable() { return !$this->closed && $this->input->isReadable(); } public function pause() { if ($this->closed) { return; } $this->paused = true; $this->input->pause(); } public function resume() { if ($this->closed) { return; } $this->paused = false; $this->input->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; // stop listening for incoming events $this->input->removeListener('data', array($this, 'handleData')); $this->input->removeListener('error', array($this, 'handleError')); $this->input->removeListener('end', array($this, 'handleEnd')); $this->input->removeListener('close', array($this, 'close')); // resume the stream to ensure we discard everything from incoming connection if ($this->paused) { $this->paused = false; $this->input->resume(); } $this->emit('close'); $this->removeAllListeners(); } /** @internal */ public function handleData($data) { $this->emit('data', array($data)); } /** @internal */ public function handleEnd() { $this->emit('end'); $this->close(); } /** @internal */ public function handleError(\Exception $e) { $this->emit('error', array($e)); } }