gistfile1.php

<?php

namespace Anse\Xmpp;

use Anse\Xmpp\Message\MessageInterface;

class Stream
{
    /**
     * @var string
     */
    protected $host;

    /**
     * @var int
     */
    protected $port = 5222;

    /**
     * @var int
     */
    protected $timeout = 30;

    /**
     * @var resource
     */
    private $socket;

    /**
     * @param string $host
     * @param int    $port
     * @param int    $timeout
     */
    public function __construct($host, $port = 5222, $timeout = 30)
    {
        $this->setHost($host);
        $this->setPort($port);
        $this->setTimeout($timeout);
    }

    public function connect()
    {
        $start = time();

        do {
            $this->socket = @stream_socket_client($this->getConnectionString(), $errno, $errstr, $this->timeout);
            if (false === $this->socket) {
                throw new \RuntimeException($errstr, $errno);
            }

            if (!$this->socket) {
                sleep($this->timeout);
            }
        } while (!$this->socket && (time() - $start) < $this->timeout);

        if ($this->socket) {
            stream_set_blocking($this->socket, 1);
        } else {
            throw new \RuntimeException('Could not connect to server.');
        }

        return $this;
    }

    /**
     * @param Message\MessageInterface $message
     * @return bool|int
     */
    public function send(MessageInterface $message)
    {
        $read   = array();
        $write  = array($this->socket);
        $except = array();

        $select = stream_select($read, $write, $except, 5);
        if ($select > 0) {
            $sent = fwrite($this->socket, $message->toXml());
            if (false !== $sent) {
                return $sent;
            }
        }

        return true;
    }

    public function hasPendingRead()
    {
        $read   = array($this->socket);
        $write  = array();
        $except = array();

        return stream_select($read, $write, $except, 5);
    }

    /**
     * @return array|bool|string
     */
    public function read()
    {
        $done = false;

        $response = '';
        while (!$done) {
            $response .= fread($this->socket, 4096);
            if ($this->hasPendingRead() == 0) {
                $done = true;
            }
        }

        return $response;
    }

    /**
     * @param $timeout
     * @return Stream
     * @throws \InvalidArgumentException
     */
    protected function setTimeout($timeout)
    {
        if (!is_integer($timeout)) {
            throw new \InvalidArgumentException('Invalid timeout provided. Expected integer, got ' . gettype($timeout));
        }

        if ($timeout < 1) {
            throw new \InvalidArgumentException('Invalid timeout provided. Should be minimum 1 second');
        }

        $this->timeout = $timeout;

        return $this;
    }

    /**
     * @param $host
     * @return Stream
     * @throws \InvalidArgumentException
     */
    protected function setHost($host)
    {
        if (!is_string($host)) {
            throw new \InvalidArgumentException('Invalid host provided. Expected string, got ' . gettype($host));
        }

        $this->host = $host;

        return $this;
    }

    /**
     * @param $port
     * @return Stream
     * @throws \InvalidArgumentException
     */
    protected function setPort($port)
    {
        if (!is_integer($port)) {
            throw new \InvalidArgumentException('Invalid port provided. Expected integer, got ' . gettype($port));
        }

        $this->port = $port;

        return $this;
    }

    /**
     * @return string
     */
    protected function getConnectionString()
    {
        return 'tcp://' . $this->host . ':' . $this->port;
    }
}