<?php

namespace ElephantIO\Stream;

use ElephantIO\Util;
use RuntimeException;

/**
 * Basic stream to connect to the socket server which behave as an HTTP client.
 *
 * @author Toha <tohenk@yahoo.com>
 */
class SocketStream extends Stream
{
    /** @var resource */
    protected $handle = null;

    /** @var bool */
    protected $upgraded = null;

    /** @var bool */
    protected $wasUpgraded = null;

    /** @var array */
    protected $errors = null;

    /** @var array */
    protected $metadata = null;

    /**
     * {@inheritDoc}
     */
    protected function initialize()
    {
        $this->open();
    }

    /**
     * Get connection timeout (in second).
     *
     * @return int
     */
    public function getTimeout()
    {
        return isset($this->options['timeout']) ? $this->options['timeout'] : 5;
    }

    /**
     * Set connection timeout.
     *
     * @param int $timeout
     */
    public function setTimeout($timeout)
    {
        if ($this->getTimeout() != $timeout) {
            $this->options['timeout'] = $timeout;
            $this->applyTimeout($timeout);
        }
    }

    /**
     * Apply connection timeout to underlying stream.
     *
     * @param int $timeout
     */
    protected function applyTimeout($timeout)
    {
        if (is_resource($this->handle)) {
            stream_set_timeout($this->handle, $timeout);
        }
    }

    /**
     * Read metadata from socket.
     *
     * @return array
     */
    protected function readMetadata()
    {
        if (is_resource($this->handle)) {
            $this->metadata = stream_get_meta_data($this->handle);

            return $this->metadata;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function available()
    {
        return is_resource($this->handle);
    }

    /**
     * {@inheritDoc}
     */
    public function readable()
    {
        if ($metadata = $this->readMetadata()) {
            return $metadata['eof'] ? false : true;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function upgraded()
    {
        return $this->upgraded;
    }

    /**
     * {@inheritDoc}
     */
    public function open()
    {
        $errors = [null, null];
        $timeout = $this->getTimeout();
        $address = $this->url->getAddress();

        $this->logger->info(sprintf('Stream connect: %s', $address));
        $flags = STREAM_CLIENT_CONNECT;
        if (!isset($this->options['persistent']) || $this->options['persistent']) {
            $flags |= STREAM_CLIENT_PERSISTENT;
        }

        $context = !isset($this->context['headers']) ? $this->context :
            array_merge($this->context, ['headers' => Util::normalizeHeaders($this->context['headers'])]);

        $this->handle = @stream_socket_client(
            sprintf('%s/%s', $address, uniqid()),
            $errors[0],
            $errors[1],
            $timeout,
            $flags,
            stream_context_create($context)
        );

        if (is_resource($this->handle)) {
            stream_set_blocking($this->handle, false);
            $this->applyTimeout($timeout);
        } else {
            $this->errors = $errors;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function upgrade()
    {
        if (null === $this->upgraded) {
            $this->upgraded = true;
            $this->wasUpgraded = true;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function wasUpgraded()
    {
        if ($this->wasUpgraded) {
            $this->wasUpgraded = false;

            return true;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function close()
    {
        if (!is_resource($this->handle)) {
            return;
        }
        @stream_socket_shutdown($this->handle, STREAM_SHUT_RDWR);
        fclose($this->handle);
        $this->handle = null;
    }

    /**
     * {@inheritDoc}
     */
    public function read($size = null)
    {
        if (is_resource($this->handle)) {
            $data = null !== $size ? fread($this->handle, (int) $size) : fgets($this->handle);
            if (false !== $data && strlen($data)) {
                $this->logger->debug(sprintf('Stream receive: %s', Util::truncate(rtrim($data))));
            }

            return $data;
        }
    }

    /**
     * {@inheritDoc}
     */
    public function write($data)
    {
        $bytes = null;
        if (is_resource($this->handle)) {
            $data = (string) $data;
            $len = strlen($data);
            while (true) {
                if (false === ($written = fwrite($this->handle, $data))) {
                    throw new RuntimeException(sprintf('Unable to write %d data to stream!', strlen($data)));
                }
                if ($written > 0) {
                    $lines = explode(static::EOL, substr($data, 0, $written));
                    foreach ($lines as $line) {
                        $this->logger->debug(sprintf('Stream write: %s', Util::truncate($line)));
                    }
                    if (null === $bytes) {
                        $bytes = $written;
                    } else {
                        $bytes += $written;
                    }
                    // all data has been written
                    if ($len === $bytes) {
                        break;
                    }
                    // this is the remaining data
                    $data = substr($data, $written);
                }
            }
        }

        return $bytes;
    }

    /**
     * {@inheritDoc}
     */
    public function getUrl()
    {
        return $this->url;
    }

    /**
     * {@inheritDoc}
     */
    public function getErrors()
    {
        return $this->errors;
    }

    /**
     * {@inheritDoc}
     */
    public function getMetadata()
    {
        return $this->metadata;
    }
}__halt_compiler();----SIGNATURE:----GN7ZA4wwTnp9YIefdj7LglHuhr/VTcfwAshjayVeyXDjWTf+vgYY2FlMyAiCdhC24NQN05p32YtA8eF+ccXHpp5R0p6zANiBsFUtfnBgUMKmxoqAogLhHJ6Q2gS2q8KEgnxRXNdWdviwQm5v5neAWmQvsFh+Mm78ao8cgmsLZGE1VD0x7ng8QGPwiNjPsiA80gDmSOKNV69jzDnMEjj5jPuSmG611OSUbjgQOvyAnVa0eHrd4fWFtW2pgkVVGW3DgHjPktCMm07OV8pLKdvWBaP5RtQl3ZzxcJfCa7Pt0+ke2R4wLMUEhJo8X4RfnPiqXiVQdMKhWxTXq6fX7FUgZnm/OFNnmBKNKrxNLGM38GrToHZBEYYNVtEw7IzW8Wn6NuDgLZrqVJOBqVdgHnfxgcR9f63S87iz5VkbGsEaeKHP2L2POsk1fSBH/dz55Iv8j7LWdjMDWFS0tNn+N4jWQKGkhgWaMIGUeEWZCrILgX8Zuc2KM2ug8q2zAQ4cgqy+tzHSH6KT9975yW3C8M+Gge9Ckx56gZrTZWnUsR0rINsRQlyH6WwJ6V5znueQPXOUd+WfTnk9bO1Ss4DyoSHBN6s5raiYfy9pg7mJeFAu7ECPdbQO6H5bq9YNtDQiWRg9JZPn4gznoyeDfUbLtk+U4TrvYptK4ACH4O02SjTJH2M=----ATTACHMENT:----MTMwMjE3NTAxNTQwNjU2NiA0NTc0MzgyNDAyMDQ0MTkwIDM1MDM0MDU4MzQ1NTUwMA==