path: root/vendor/packaged/thrift/src/Transport/TSocket.php

<?php
/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 *
 * @package thrift.transport
 */

namespace Thrift\Transport;

use Thrift\Exception\TException;
use Thrift\Exception\TTransportException;
use Thrift\Factory\TStringFuncFactory;

/**
 * Sockets implementation of the TTransport interface.
 *
 * @package thrift.transport
 */
class TSocket extends TTransport
{
    /**
     * Handle to PHP socket
     *
     * @var resource
     */
    protected $handle_ = null;

    /**
     * Remote hostname
     *
     * @var string
     */
    protected $host_ = 'localhost';

    /**
     * Remote port
     *
     * @var int
     */
    protected $port_ = '9090';

    /**
     * Send timeout in seconds.
     *
     * Combined with sendTimeoutUsec this is used for send timeouts.
     *
     * @var int
     */
    protected $sendTimeoutSec_ = 0;

    /**
     * Send timeout in microseconds.
     *
     * Combined with sendTimeoutSec this is used for send timeouts.
     *
     * @var int
     */
    protected $sendTimeoutUsec_ = 100000;

    /**
     * Recv timeout in seconds
     *
     * Combined with recvTimeoutUsec this is used for recv timeouts.
     *
     * @var int
     */
    protected $recvTimeoutSec_ = 0;

    /**
     * Recv timeout in microseconds
     *
     * Combined with recvTimeoutSec this is used for recv timeouts.
     *
     * @var int
     */
    protected $recvTimeoutUsec_ = 750000;

    /**
     * Persistent socket or plain?
     *
     * @var bool
     */
    protected $persist_ = false;

    /**
     * Debugging on?
     *
     * @var bool
     */
    protected $debug_ = false;

    /**
     * Debug handler
     *
     * @var mixed
     */
    protected $debugHandler_ = null;

    /**
     * Socket constructor
     *
     * @param string $host Remote hostname
     * @param int $port Remote port
     * @param bool $persist Whether to use a persistent socket
     * @param string $debugHandler Function to call for error logging
     */
    public function __construct(
        $host = 'localhost',
        $port = 9090,
        $persist = false,
        $debugHandler = null
    ) {
        $this->host_ = $host;
        $this->port_ = $port;
        $this->persist_ = $persist;
        $this->debugHandler_ = $debugHandler ? $debugHandler : 'error_log';
    }

    /**
     * @param resource $handle
     * @return void
     */
    public function setHandle($handle)
    {
        $this->handle_ = $handle;
        stream_set_blocking($this->handle_, false);
    }

    /**
     * Sets the send timeout.
     *
     * @param int $timeout Timeout in milliseconds.
     */
    public function setSendTimeout($timeout)
    {
        $this->sendTimeoutSec_ = floor($timeout / 1000);
        $this->sendTimeoutUsec_ =
            ($timeout - ($this->sendTimeoutSec_ * 1000)) * 1000;
    }

    /**
     * Sets the receive timeout.
     *
     * @param int $timeout Timeout in milliseconds.
     */
    public function setRecvTimeout($timeout)
    {
        $this->recvTimeoutSec_ = floor($timeout / 1000);
        $this->recvTimeoutUsec_ =
            ($timeout - ($this->recvTimeoutSec_ * 1000)) * 1000;
    }

    /**
     * Sets debugging output on or off
     *
     * @param bool $debug
     */
    public function setDebug($debug)
    {
        $this->debug_ = $debug;
    }

    /**
     * Get the host that this socket is connected to
     *
     * @return string host
     */
    public function getHost()
    {
        return $this->host_;
    }

    /**
     * Get the remote port that this socket is connected to
     *
     * @return int port
     */
    public function getPort()
    {
        return $this->port_;
    }

    /**
     * Tests whether this is open
     *
     * @return bool true if the socket is open
     */
    public function isOpen()
    {
        return is_resource($this->handle_);
    }

    /**
     * Connects the socket.
     */
    public function open()
    {
        if ($this->isOpen()) {
            throw new TTransportException('Socket already connected', TTransportException::ALREADY_OPEN);
        }

        if (empty($this->host_)) {
            throw new TTransportException('Cannot open null host', TTransportException::NOT_OPEN);
        }

        if ($this->port_ <= 0) {
            throw new TTransportException('Cannot open without port', TTransportException::NOT_OPEN);
        }

        if ($this->persist_) {
            $this->handle_ = @pfsockopen(
                $this->host_,
                $this->port_,
                $errno,
                $errstr,
                $this->sendTimeoutSec_ + ($this->sendTimeoutUsec_ / 1000000)
            );
        } else {
            $this->handle_ = @fsockopen(
                $this->host_,
                $this->port_,
                $errno,
                $errstr,
                $this->sendTimeoutSec_ + ($this->sendTimeoutUsec_ / 1000000)
            );
        }

        // Connect failed?
        if ($this->handle_ === false) {
            $error = 'TSocket: Could not connect to ' .
                $this->host_ . ':' . $this->port_ . ' (' . $errstr . ' [' . $errno . '])';
            if ($this->debug_) {
                call_user_func($this->debugHandler_, $error);
            }
            throw new TException($error);
        }

        if (function_exists('socket_import_stream') && function_exists('socket_set_option')) {
            $socket = socket_import_stream($this->handle_);
            socket_set_option($socket, SOL_TCP, TCP_NODELAY, 1);
        }
    }

    /**
     * Closes the socket.
     */
    public function close()
    {
        @fclose($this->handle_);
        $this->handle_ = null;
    }

    /**
     * Read from the socket at most $len bytes.
     *
     * This method will not wait for all the requested data, it will return as
     * soon as any data is received.
     *
     * @param int $len Maximum number of bytes to read.
     * @return string Binary data
     */
    public function read($len)
    {
        $null = null;
        $read = array($this->handle_);
        $readable = @stream_select(
            $read,
            $null,
            $null,
            $this->recvTimeoutSec_,
            $this->recvTimeoutUsec_
        );

        if ($readable > 0) {
            $data = fread($this->handle_, $len);
            if ($data === false) {
                throw new TTransportException('TSocket: Could not read ' . $len . ' bytes from ' .
                    $this->host_ . ':' . $this->port_);
            } elseif ($data == '' && feof($this->handle_)) {
                throw new TTransportException('TSocket read 0 bytes');
            }

            return $data;
        } elseif ($readable === 0) {
            throw new TTransportException('TSocket: timed out reading ' . $len . ' bytes from ' .
                $this->host_ . ':' . $this->port_);
        } else {
            throw new TTransportException('TSocket: Could not read ' . $len . ' bytes from ' .
                $this->host_ . ':' . $this->port_);
        }
    }

    /**
     * Write to the socket.
     *
     * @param string $buf The data to write
     */
    public function write($buf)
    {
        $null = null;
        $write = array($this->handle_);

        // keep writing until all the data has been written
        while (TStringFuncFactory::create()->strlen($buf) > 0) {
            // wait for stream to become available for writing
            $writable = @stream_select(
                $null,
                $write,
                $null,
                $this->sendTimeoutSec_,
                $this->sendTimeoutUsec_
            );
            if ($writable > 0) {
                // write buffer to stream
                $written = fwrite($this->handle_, $buf);
                if ($written === -1 || $written === false) {
                    throw new TTransportException(
                        'TSocket: Could not write ' . TStringFuncFactory::create()->strlen($buf) . ' bytes ' .
                        $this->host_ . ':' . $this->port_
                    );
                }
                // determine how much of the buffer is left to write
                $buf = TStringFuncFactory::create()->substr($buf, $written);
            } elseif ($writable === 0) {
                throw new TTransportException(
                    'TSocket: timed out writing ' . TStringFuncFactory::create()->strlen($buf) . ' bytes from ' .
                    $this->host_ . ':' . $this->port_
                );
            } else {
                throw new TTransportException(
                    'TSocket: Could not write ' . TStringFuncFactory::create()->strlen($buf) . ' bytes ' .
                    $this->host_ . ':' . $this->port_
                );
            }
        }
    }

    /**
     * Flush output to the socket.
     *
     * Since read(), readAll() and write() operate on the sockets directly,
     * this is a no-op
     *
     * If you wish to have flushable buffering behaviour, wrap this TSocket
     * in a TBufferedTransport.
     */
    public function flush()
    {
        // no-op
    }
}