Add the message body if the stream is seekable.

@param MessageInterface $request
@param string           $message

@return string
489372
private function addBody(MessageInterface $request, $message)
    {
        $stream = $request->getBody();
        if (!$stream->isSeekable() || 0 === $this->maxBodyLength) {
            // Do not read the stream
            return $message."\n";
        }

        if (null === $this->maxBodyLength) {
            $message .= "\n".$stream->__toString();
        } else {
            $message .= "\n".mb_substr($stream->__toString(), 0, $this->maxBodyLength);
        }

        $stream->rewind();

        return $message;
    }
472666
public function onMessage(MessageEvent $event){
        if($event->message instanceof HttpRequest){
            return $this->processHandshake($event);
        }
        
        if(!($event->message instanceof Request)){
            Yii::error("message for ".get_class($this).'::onMessage must be StreamRequest');
            return ;
        }
        
        return $this->processStream($event);
    }
1035479
private function addDebugToolBar(StreamInterface $body){
        $newBody    = new \Fratily\Http\Message\Stream\MemoryStream();

        $body->rewind();
        $newBody->write($this->debugbar->embed($body->getContents()));

        return $newBody;
    }
304469
private function getFromCache($key, MessageInterface $message)
    {
        if ($this->cache) {
            $item = $this->cache->getItem($key);

            if ($item->isHit()) {
                list($headers, $body) = $item->get();

                foreach ($headers as $name => $header) {
                    $message = $message->withHeader($name, $header);
                }

                $message->getBody()->write($body);

                return $message;
            }
        }

        return false;
    }
1184609
public function compose(MessageInterface $message, $params = [])
    {
        $this->message = $message;

        if (is_array($this->viewName)) {
            if (isset($this->viewName['html'])) {
                $html = $this->render($this->viewName['html'], $params, $this->htmlLayout);
            }
            if (isset($this->viewName['text'])) {
                $text = $this->render($this->viewName['text'], $params, $this->textLayout);
            }
        } else {
            $html = $this->render($this->viewName, $params, $this->htmlLayout);
        }

        if (isset($html)) {
            $this->message->setHtmlBody($html);
        }
        if (isset($text)) {
            $this->message->setTextBody($text);
        } elseif (isset($html)) {
            if (preg_match('~<body[^>]*>(.*?)</body>~is', $html, $match)) {
                $html = $match[1];
            }
            // remove style and script
            $html = preg_replace('~<((style|script))[^>]*>(.*?)</\1>~is', '', $html);
            // strip all HTML tags and decode HTML entities
            $text = html_entity_decode(strip_tags($html), ENT_QUOTES | ENT_HTML5);
            // improve whitespace
            $text = preg_replace("~^[ \t]+~m", '', trim($text));
            $text = preg_replace('~\R\R+~mu', "\n\n", $text);
            $this->message->setTextBody($text);
        }
    }
108652
public function withBody(StreamInterface $body)
    {
        $request = clone $this;
        if (method_exists($request, 'setContent')) {
            $request->setContent($body);
        } else {
            $request->content = $body;
        }

        return $request;
    }
891881
public function modify(MessageInterface $message) : MessageInterface
    {
        /** @var MessageModifierInterface $modifier */
        foreach ($this->collection as $modifier) {
            $message = $modifier->modify($message);
        }

        return $message;
    }
714713
public function createRequest(PushMessage $message) : RequestInterface
    {
        $host = $message->getPushSubscription()->getEndpoint()->getHost();
        $pushService = $this->pushServiceRegistry->getPushService($host);

        return $pushService->createRequest($message);
    }
1042260
public function addMessage(Message $message)
    {
        $type = $message->getType();
        if (!array_key_exists($type, $this->messages)) {
            $this->messages[$type] = [];
        }

        if ($message->getId() !== null) {
            $this->messages[$type][$message->getId()] = $message;
        } else {
            $this->messages[$type][] = $message;
        }

        return $this;
    }
761400
public function addMessage($message) {
        $migrationName = get_class($this);
        $this->messages[] = sprintf("%s:  %s\n", $migrationName, $message);
    }
806625
public static function fromStream(StreamInterface $stream)
    {
        if (! $stream->isReadable() || ! $stream->isSeekable()) {
            throw new InvalidArgumentException('Message stream must be both readable and seekable');
        }

        $stream->rewind();

        list($version, $status, $reasonPhrase) = self::getStatusLine($stream);
        list($headers, $body)                  = self::splitStream($stream);

        return (new Response($body, $status, $headers))
            ->withProtocolVersion($version)
            ->withStatus($status, $reasonPhrase);
    }
946570
protected function headerToArray(MessageInterface $message, $header)
    {
        $lines = explode(Http::NEW_LINE, $header);

        foreach ($lines as $line) {
            $parts = explode(':', $line, 2);

            if (isset($parts[0]) && isset($parts[1])) {
                $key   = $parts[0];
                $value = substr($parts[1], 1);

                $message->addHeader($key, $value);
            }
        }
    }
60915
public function save(Message $message) : ?Message
    {
        $aggregate = $message->getHeaders()->get(AggregateMessage::AGGREGATE_OBJECT);

        $this->aggregateRepository->save($message->getHeaders()->get(AggregateMessage::CALLING_MESSAGE), $aggregate);

        if ($message->getHeaders()->get(AggregateMessage::IS_FACTORY_METHOD)) {
            $aggregate = $message->getHeaders()->get(AggregateMessage::AGGREGATE_OBJECT);
            $aggregateIds = [];

            foreach ($message->getHeaders()->get(AggregateMessage::AGGREGATE_ID) as $aggregateIdName => $aggregateIdValue) {
                $aggregateIds[$aggregateIdName] = $this->propertyReaderAccessor->getPropertyValue(PropertyPath::createWith($aggregateIdName), $aggregate);
            }

            $payload = count($aggregateIds) === 1 ? array_shift($aggregateIds) : $aggregateIds;

            return
                MessageBuilder::fromMessage($message)
                    ->setPayload($payload)
                    ->build();
        }

        return $message;
    }
1222006
protected function getBody(\Psr\Http\Message\MessageInterface $message)
    {
        $body = trim($message->getBody());

        if ($message->getBody()->getSize() === 0 || empty($body)) {
            if ($message instanceof \Psr\Http\Message\ResponseInterface) {
                return 'No response body returned';
            }
            return 'No request body sent';
        }
        $result = json_decode($body);
        if ($result !== null) {
            return json_encode($result, JSON_PRETTY_PRINT);
        }

        return $body;
    }
928325
public function sendMessage($message)
    {
        $message = sprintf("%s\r\n", $message);
        fwrite($this->getSocketStream(), $message, strlen($message));
        return $this->receiveMessage();
    }
668929
public function createStream($body = null)
    {
        if ($body instanceof StreamInterface)
        {
            return $body;
        }

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

        return Stream::create($body ?? '');
    }
163434
public function addMessage($message) {
        $this->message = trim(trim($this->message).NL.$message);
        return $this;
    }