Blocking socket read.

@param resource $socket  The socket to read from
@param float    $timeout The timeout for the read

@throws \MHlavac\Gearman\Exception on timeouts

@return array
768514
public static function blockingRead($socket, $timeout = 500.0)
    {
        static $cmds = array();

        $tv_sec = floor(($timeout % 1000));
        $tv_usec = ($timeout * 1000);

        $start = microtime(true);
        while (count($cmds) == 0) {
            if (((microtime(true) - $start) * 1000) > $timeout) {
                throw new Exception('Blocking read timed out');
            }

            $write = null;
            $except = null;
            $read = array($socket);

            socket_select($read, $write, $except, $tv_sec, $tv_usec);
            foreach ($read as $s) {
                $cmds[] = self::read($s);
            }
        }

        return array_shift($cmds);
    }
188223
public function setTimeout($timeout)
    {
        $timeout = is_scalar($timeout) ? (int) $timeout : 0;

        return $timeout < 1
            ? $this->timeout
            : $this->timeout = $timeout;
    }
913700
protected function socket_receive($socket)
	{
		// initially read to the length of the header size, then
		// expand to read more
		$bytes_total = self::SOCKET_HEADER_SIZE;
		$bytes_read = 0;
		$have_header = false;
		$socket_message = '';
		while ($bytes_read < $bytes_total)
		{
			$read = @socket_read($socket, $bytes_total - $bytes_read);
			if ($read === false)
			{
				$this->log('socket_receive error: ' . socket_strerror(socket_last_error()), self::LOG_LEVEL_CRIT);
				return false;
			}

			// blank socket_read means done
			if ($read == '')
			{
				break;
			}

			$bytes_read += strlen($read);
			$socket_message .= $read;

			if (!$have_header && $bytes_read >= self::SOCKET_HEADER_SIZE)
			{
				$have_header = true;
				list($bytes_total) = array_values(unpack('N', $socket_message));
				$bytes_read = 0;
				$socket_message = '';
			}
		}

		$message = @unserialize($socket_message);

		return $message;
	}
912969
public function connect($timeout = self::TIMEOUT, $test = false)
    {
        if ($this->socket) {
            return $this;
        }

        $host = $this->host;

        if ($this->ssl) {
            $host = 'ssl://' . $host;
        }

        $errno  =  0;
        $errstr = '';

        $this->socket = @fsockopen($host, $this->port, $errno, $errstr, $timeout);

        if (!$this->socket) {
            //throw exception
            Exception::instance()
                ->setMessage(Exception::SERVER_ERROR)
                ->addVariable($host.':'.$this->port)
                ->trigger();
        }

        if (strpos($this->getLine(), '* OK') === false) {
            $this->disconnect();
            //throw exception
            Exception::instance()
                ->setMessage(Exception::SERVER_ERROR)
                ->addVariable($host.':'.$this->port)
                ->trigger();
        }

        if ($this->tls) {
            $this->send('STARTTLS');
            if (!stream_socket_enable_crypto($this->socket, true, STREAM_CRYPTO_METHOD_TLS_CLIENT)) {
                $this->disconnect();
                //throw exception
                Exception::instance()
                ->setMessage(Exception::TLS_ERROR)
                ->addVariable($host.':'.$this->port)
                ->trigger();
            }
        }

        if ($test) {
            fclose($this->socket);

            $this->socket = null;
            return $this;
        }

        //login
        $result = $this->call('LOGIN', $this->escape($this->username, $this->password));

        if (!is_array($result) || strpos(implode(' ', $result), 'OK') === false) {
            $this->disconnect();
            //throw exception
            Exception::instance(Exception::LOGIN_ERROR)->trigger();
        }

        return $this;
    }
405758
public function acquireLock($name, $timeout = null)
    {
        $blocking = $timeout === null;
        $start = microtime(true);
        $end = $start + $timeout / 1000;
        $locked = false;
        while (!(empty($this->locks[$name]) && $locked = $this->getLock($name, $blocking)) && ($blocking || ($timeout > 0 && microtime(true) < $end))) {
            usleep(static::USLEEP_TIME);
        }

        if ($locked) {
            $this->locks[$name] = true;

            return true;
        }

        return false;
    }
1284053
public function openSocket(int $port, int $timeout = 5, callable $openWith = null): Stream
    {
        $socket = new Socket($this->ip, $port, null);
        if (null !== $openWith) {
            $socket->openWith($openWith);
        }

        return $socket->connect()->setTimeout($timeout);
    }
1094715
public function receive($timeout = 0)
    {
        $startAt = microtime(true);

        while (true) {
            $message = $this->receiveMessage();

            if ($message) {
                return $message;
            }

            if ($timeout && (microtime(true) - $startAt) >= $timeout) {
                return;
            }

            usleep($this->pollingInterval);

            if ($timeout && (microtime(true) - $startAt) >= $timeout) {
                return;
            }
        }
    }
233779
public function block($timeout, $callback = null)
    {
        $starting = time();

        $id = Str::random(20);

        while (! $slot = $this->acquire($id)) {
            if (time() - $timeout >= $starting) {
                throw new LimiterTimeoutException;
            }

            usleep(250 * 1000);
        }

        if (is_callable($callback)) {
            try {
                return tap($callback(), function () use ($slot, $id) {
                    $this->release($slot, $id);
                });
            } catch (Exception $exception) {
                $this->release($slot, $id);

                throw $exception;
            }
        }

        return true;
    }
1109723
public function checkAtOnce(string $addr, int $port = 0, int $timeout = 3): Result {
        try {
            $socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
            if (!$socket) {
                $errno = socket_last_error();
                throw new \Exception("socket create error:" . socket_strerror($errno), $errno);
            }
            socket_set_option($socket, SOL_SOCKET, SO_RCVTIMEO, array('sec' => $timeout, 'usec' => 0));
            socket_set_option($socket, SOL_SOCKET, SO_SNDTIMEO, array('sec' => $timeout, 'usec' => 0));
            $ok = @socket_connect($socket, $addr, $port);
            $err = "";
            if (!$ok) {
                $err = socket_strerror(socket_last_error());
            }
            socket_close($socket);
        } catch (\Exception $e) {
            $ok = false;
            $err = $e->getMessage();
        }
        return new Result(static::class, $addr, $port, $ok, $err);
    }
300671
public function block($timeout, $callback = null)
    {
        $starting = time();

        while (! $this->acquire()) {
            if (time() - $timeout >= $starting) {
                throw new LimiterTimeoutException;
            }

            usleep(750 * 1000);
        }

        if (is_callable($callback)) {
            $callback();
        }

        return true;
    }
864943
private function waitForStreamActivity($timeout)
    {
        $read  = $this->readStreams;
        $write = $this->writeStreams;

        if ($this->streamSelect($read, $write, $timeout) === false)
        {
            return;
        }

        foreach ($read as $stream)
        {
            $key = (int) $stream;

            if (isset($this->readListeners[$key]))
            {
                $callable = $this->readListeners[$key];
                $callable($stream, $this);
            }
        }

        foreach ($write as $stream)
        {
            $key = (int) $stream;

            if (isset($this->writeListeners[$key]))
            {
                $callable = $this->writeListeners[$key];
                $callable($stream, $this);
            }
        }
    }
481653
public function connect($host, $port, $timeout = 1)
    {
        $this->socket = @stream_socket_client(
            "tcp://$host:$port",
            $errorNumber,
            $errorMessage,
            $timeout,
            STREAM_CLIENT_CONNECT,
            stream_context_create($this->options)
        );

        return (bool) $this->socket;
    }
224357
public function setRequestTimeout($timeout)
    {
        if (!is_int($timeout) || $timeout < 0) {
            throw new Exception("Invalid value supplied for request timeout: $timeout");
        }

        $this->requestTimeout = $timeout;
        return $this;
    }
694498
public function connect($timeout = 0.5)
    {
        if (!extension_loaded('swoole'))
        {
            throw new \Exception('swoole extension is required for WebSocketAsync');
        }

        //没有onMessage
        if (!$this->_callback('message'))
        {
            throw new \Exception('no message event callback.');
        }

        $type = SWOOLE_TCP;
        if ($this->ssl)
        {
            $type |= SWOOLE_SSL;
        }

        $this->socket = new \swoole_client($type, SWOOLE_SOCK_ASYNC);
        if ($this->ssl_key_file)
        {
            $this->socket->set(array(
                'ssl_key_file' => $this->ssl_key_file,
                'ssl_cert_file' => $this->ssl_cert_file
            ));
        }

        $this->socket->on("connect", array($this, 'onConnect'));
        $this->socket->on("receive", array($this, 'onReceive'));
        $this->socket->on("error", array($this, 'onError'));
        $this->socket->on("close", array($this, 'onClose'));

        $this->socket->connect($this->host, $this->port, $timeout);
    }
1017221
public function openSocket(int $timeout = 5, callable $openWith = null): Stream
    {
        $socket = $this->createSocket();
        if (null !== $openWith) {
            $socket->openWith($openWith);
        }

        return $socket->connect()->setTimeout($timeout);
    }
1248783
protected Socket acceptSocket(int timeout)
    {
        try
        {
            Socket s = null;

            if (_listen != null)
            {
                if (_soTimeOut != timeout)
                {
                    _soTimeOut = timeout;
                    _listen.setSoTimeout(_soTimeOut);
                }

                s = _listen.accept();

                try
                {
                    if (getMaxIdleTimeMs() >= 0) s.setSoTimeout(getMaxIdleTimeMs());
                    if (_lingerTimeSecs >= 0)
                        s.setSoLinger(true, _lingerTimeSecs);
                    else
                        s.setSoLinger(false, 0);
                }
                catch (Exception e)
                {
                    LogSupport.ignore(log, e);
                }
            }
            return s;
        }
        catch (java.net.SocketException e)
        {
            // TODO - this is caught and ignored due strange
            // exception from linux java1.2.v1a
            LogSupport.ignore(log, e);
        }
        catch (InterruptedIOException e)
        {
            LogSupport.ignore(log, e);
        }
        catch (IOException e)
        {
            log.warn(LogSupport.EXCEPTION, e);
        }
        return null;
    }
683538
protected function makeSocket($timeout = 1)
    {
        return fsockopen(
            'tcp://' . $this->host,
            $this->port,
            $errno,
            $errstr,
            $timeout
        );
    }