@param EventStreamInterface $eventStream

@return CryptoEventStream
515091
public function createDecryptingStream(EventStreamInterface $eventStream)
    {
        return new CryptoEventStream(
            $eventStream,
            $this->cryptos,
            $this->encryptionKey,
            CryptoEventStream::DECRYPT
        );
    }
95855
public function withBody(StreamInterface $body)
    {
        $message = clone $this;
        $stream = $message->stream();
        $stream->close();
        $stream->add($body);
        return $message;
    }
469347
public static function createSingleStream($streamName = 'event_stream', $withCausationColumns = false)
    {
        Schema::create($streamName, function (Blueprint $eventStream) use ($streamName, $withCausationColumns) {
            // UUID4 of the event
            $eventStream->char('event_id', 36);
            // Version of the aggregate after event was recorded
            $eventStream->integer('version', false, true);
            // Name of the event
            $eventStream->string('event_name', 100);
            // Event payload
            $eventStream->text('payload');
            // DateTime ISO8601 + microseconds UTC stored as a string e.g. 2016-02-02T11:45:39.000000
            $eventStream->char('created_at', 26);
            // UUID4 of linked aggregate
            $eventStream->char('aggregate_id', 36);
            // Class of the linked aggregate
            $eventStream->string('aggregate_type', 150);

            if ($withCausationColumns) {
                // UUID4 of the command which caused the event
                $eventStream->char('causation_id', 36);
                // Name of the command which caused the event
                $eventStream->string('causation_name', 100);
            }
            $eventStream->primary('event_id');
            // Concurrency check on database level
            $eventStream->unique(['aggregate_id', 'aggregate_type', 'version'], $streamName . '_m_v_uix');
        });
    }
566130
private function limitEventStreamToPlayhead(DomainEventStreamInterface $eventStream, $playhead)
    {
        return array_filter(
            iterator_to_array($eventStream),
            function (DomainMessage $message) use ($playhead) {
                return intval($message->getPlayhead()) < $playhead;
            }
        );
    }
507138
private function decorateWithHashes(Stream $stream, array &$data)
    {
        // Decorate source with a hashing stream
        $hash = new PhpHash('sha256');
        return new HashingStream($stream, $hash, function ($result) use (&$data) {
            $data['ContentSHA256'] = bin2hex($result);
        });
    }
403316
public function publish(DomainEventStream $domainMessages)
    {
        $this->transactionManager->beginTransaction();
        try {
            $this->eventBus->publish($domainMessages);
            $this->transactionManager->commit();
        } catch (\Exception $e) {
            $this->transactionManager->rollback();
            throw $e;
        }
    }
1144588
private function decorateStream($encoding, StreamInterface $stream)
    {
        if ('chunked' == strtolower($encoding)) {
            return new DechunkStream($stream);
        }

        if ('compress' == strtolower($encoding)) {
            return new DecompressStream($stream);
        }

        if ('deflate' == strtolower($encoding)) {
            return new InflateStream($stream);
        }

        if ('gzip' == strtolower($encoding)) {
            return new GzipDecodeStream($stream);
        }

        return false;
    }
624514
public function decryptStream($source, $sin, VirgilPrivateKey $recipientPrivateKey)
    {
        try {
            $cipher = $this->cryptoService->createStreamCipher();
            $cipherInputOutput = $cipher->createInputOutput($source, $sin);

            $cipher->decryptWithKey(
                $cipherInputOutput,
                $recipientPrivateKey->getReceiverID(),
                $recipientPrivateKey->getValue()
            );

        } catch (Exception $exception) {
            throw new VirgilCryptoException($exception->getMessage());
        }
    }
1043532
protected function setStream($type, StreamInterface $stream)
    {
        $this->streams[$type] = $stream;

        $callback = array($this, 'on'.ucfirst($type).'Message');

        $stream->addListener(
            StreamInterface::MESSAGE,
            $this->createEventListenerForStream($callback)
        );

        return $this;
    }
1121860
public function setStream($stream)
    {
        parent::setStream($stream);
        foreach ($this->getStreamWriters() as $writer) {
            $writer->setStream($stream);
        }

        return $this;
    }
19921
public static function stream($body = null): StreamInterface
    {
        if ($body instanceof StreamInterface) {
            return $body;
        }

        if (\is_resource($body)) {
            return new Stream($body);
        }

        $resource = \fopen('php://temp', 'rw+b');
        $stream = new Stream($resource);
        if ($body) {
            $stream->write($body);
        }

        return $stream;
    }
378657
public function parse(StreamInterface $stream)
    {
        $partBuilder = $this->read($stream);
        return $partBuilder->createMessagePart($stream);
    }
274019
private function writePayloadOnStreamForGivenToken($payload, $stream, $token)
    {
        $binaryMessage = sprintf('%s%s%s%s%s',
            chr(0),
            pack('n', 32),
            pack('H*', str_replace(' ', '', $token)),
            pack('n', strlen($payload)),
            $payload
        );

        fwrite($stream, $binaryMessage, strlen($binaryMessage));
    }
460032
public function newInstance(PartBuilder $partBuilder, StreamInterface $messageStream = null)
    {
        $partStream = null;
        $contentStream = null;
        if ($messageStream !== null) {
            $partStream = $this->streamFactory->getLimitedPartStream($messageStream, $partBuilder);
            $contentStream = $this->streamFactory->getLimitedContentStream($messageStream, $partBuilder);
        }
        return new NonMimePart(
            $this->partStreamFilterManagerFactory->newInstance(),
            $this->streamFactory,
            $partStream,
            $contentStream
        );
    }
891468
public function setInputStream($stream)
    {
        $stream = new Stream($stream ?: STDIN);
        $stream = FnStream::decorate($stream, [
            'read' => function ($length) use ($stream) {
                $ret = $stream->read($length);
                $this->onRead($ret);

                return $ret;
            },
        ]);
        $stream = StreamWrapper::getResource($stream);

        parent::setInputStream($stream);
    }
177826
public function getBodyWithHeaders(MessageInterface $stream)
    {
        $result = sprintf(
            '%s%s%s',
            $this->getString($stream->getHeaders()),
            PHP_EOL,
            $stream->getBody()->getContents()
        );

        return $result;
    }
510626
public function initFromStream($stream)
    {
        $stream = Factory::stream($stream);

        try {
            $offset = $stream->tell();
        } catch (\RuntimeException $e) {
            $offset = 0;
        }

        $shouldAndCanSeek = $offset !== 0 && $stream->isSeekable();

        if ($shouldAndCanSeek) {
            $stream->rewind();
        }

        try {
            $data = $stream->getContents();
        } catch (\RuntimeException $e) {
            $data = null;
        }

        if ($shouldAndCanSeek) {
            $stream->seek($offset);
        }

        if ($data) {
            return $this->initFromBinary($data);
        }

        throw new Exception\NotReadableException(
            "Unable to init from given stream"
        );
    }