Attempt to write the request to the socket, wait for response if debug
mode is enabled.
@param  stream  $socket the handle for the socket
@param  string  $req    request body
@param  boolean $retry
@return boolean $success
763875
private function makeRequest($socket, $req, $retry = true) {
    $bytes_written = 0;
    $bytes_total = strlen($req);
    $closed = false;

    // Retries with exponential backoff until success
    $backoff = 100;   // Set initial waiting time to 100ms

    while (true) {
      // Send request to server
      while (!$closed && $bytes_written < $bytes_total) {
        try {
          // Since we're try catch'ing prevent PHP logs.
          $written = @fwrite($socket, substr($req, $bytes_written));
        } catch (Exception $e) {
          $this->handleError($e->getCode(), $e->getMessage());
          $closed = true;
        }
        if (!isset($written) || !$written) {
          $closed = true;
        } else {
          $bytes_written += $written;
        }
      }

      // Get response for request
      $statusCode = 0;
      $errorMessage = "";

      if (!$closed) {
        $res = $this->parseResponse(fread($socket, 2048));
        $statusCode = (int)$res["status"];
        $errorMessage = $res["message"];
      }
      fclose($socket);

      // If status code is 200, return true
      if (200 == $statusCode) {
        return true;
      }

      // If status code is greater than 500 and less than 600, it indicates server error
      // Error code 429 indicates rate limited.
      // Retry uploading in these cases.
      if (($statusCode >= 500 && $statusCode <= 600) || 429 == $statusCode || 0 == $statusCode) {
        if ($backoff >= $this->maximum_backoff_duration) {
          break;
        }

        usleep($backoff * 1000);
      } elseif ($statusCode >= 400) {
        if ($this->debug()) {
          $this->handleError($res["status"], $res["message"]);
        }

        break;
      }

      // Retry uploading...
      $backoff *= 2;
      $socket = $this->createSocket();
    }

    return $success;
  }
921884
public void writeSocket(@NotNull final WireOut wire, boolean reconnectOnFailure) {
        assert outBytesLock().isHeldByCurrentThread();

        try {
            assert wire.startUse();
            @Nullable SocketChannel clientChannel = this.clientChannel;

            // wait for the channel to be non null
            if (clientChannel == null) {
                if (!reconnectOnFailure) {
                    return;
                }
                final byte[] bytes = wire.bytes().toByteArray();
                assert wire.endUse();
                condition.await(10, TimeUnit.SECONDS);
                assert wire.startUse();
                wire.clear();
                wire.bytes().write(bytes);
            }

            writeSocket1(wire, this.clientChannel);
        } catch (ClosedChannelException e) {
            closeSocket();
            Jvm.pause(500);
            if (reconnectOnFailure)
                throw new ConnectionDroppedException(e);

        } catch (IOException e) {
            if (!"Broken pipe".equals(e.getMessage()))
                Jvm.warn().on(getClass(), e);
            closeSocket();
            Jvm.pause(500);
            throw new ConnectionDroppedException(e);

        } catch (ConnectionDroppedException e) {
            closeSocket();
            Jvm.pause(500);
            throw e;

        } catch (Exception e) {
            Jvm.warn().on(getClass(), e);
            closeSocket();
            Jvm.pause(500);
            throw new ConnectionDroppedException(e);

        } finally {
            assert wire.endUse();
        }
    }
441593
public function connect($retry = false) {
		return $this->_backoff(
			function() { return $this->_connection->connect(); },
			$retry ? false : 1
		);
	}
476260
def write(s)
      begin
        retry_id ||= 0
        @socket.send(:write, s)
      rescue => e
        if (retry_id += 1) < @retries
          connect
          retry
        else
          raise e
        end
      end
    end
147601
final public function socketWrite($message = '', $length = null)
	{
		return @socket_write(
			$this->socket,
			"{$message}",
			is_int($length) ? $length : strlen($message)
		);
	}
1156149
def wait_for_socket
      return unless Jets.rack?

      retries = 0
      max_retries = 30 # 15 seconds at a delay of 0.5s
      delay = 0.5
      if ENV['C9_USER'] # overrides for local testing
        max_retries = 3
        delay = 3
      end
      begin
        server = TCPSocket.new('localhost', 9292)
        server.close
      rescue Errno::ECONNREFUSED
        puts "Unable to connect to localhost:9292. Delay for #{delay} and will try to connect again."  if ENV['JETS_DEBUG']
        sleep(delay)
        retries += 1
        if retries < max_retries
          retry
        else
          puts "Giving up on trying to connect to localhost:9292"
          return false
        end
      end
      puts "Connected to localhost:9292 successfully"
      true
    end
261264
protected function _writeStream($buffer) {
        $fd = $this->socket_;
        $total = TStringFuncFactory::create()->strlen($buffer);

        // Early out on the trivial case
        if ($total <= 0) {
            return;
        }

        $failed = FALSE;
        $sent = 0;

        while (!$failed && $sent < $total) {

            if (!is_resource($fd)) {
                $failed = TRUE;
            }

            try {
                // Supress any error messages as it is considered part of normal
                // operation for the write to fail on a broken pipe or timeout
                $written = @fwrite($fd, $buffer, self::MAX_BYTES_PER_WRITE);

                if ($written > 0){
                    $sent += $written;
                    $buffer = substr($buffer, $written);

                } else if ($written === FALSE) {
                    if ($this->debug_) {
                        error_log("Write failed.");
                    }
                    $failed = TRUE;

                } else if ($written === 0) {
                    if ($this->debug_) {
                        error_log("Zero bytes written to socket. sent=$sent total=$total");
                    }
                    $failed = TRUE;
                } else {
                    if ($this->debug_) {
                        error_log("Unexpected fwrite return value '$written'");
                    }
                    $failed = TRUE;
                }

                if ($this->debug_) {
                    error_log("Written = $written bytes. Sent $sent of $total.");
                }

            } catch (Exception $e) {
                if ($this->debug_) {
                    error_log($e);
                }
                $failed = TRUE;
            }
        }

        if ($failed) {
            $this->_closeSocket();
        }
    }
458915
public synchronized void sendRetry(long retry, EventCallback callback) {

        if (open == 0 || shutdown) {
            if (callback != null) {
                callback.failed(this, null, null, null, new ClosedChannelException());
            }
            return;
        }
        queue.add(new SSEData(retry, callback));
        sink.getIoThread().execute(new Runnable() {
            @Override
            public void run() {
                synchronized (ServerSentEventConnection.this) {
                    if (pooled == null) {
                        fillBuffer();
                        writeListener.handleEvent(sink);
                    }
                }
            }
        });
    }
437483
public function onRequest($req, $res)
    {
        $this->app->event->trigger('swoole.request');

        $this->resetOnRequest();

        /** @var Sandbox $sandbox */
        $sandbox = $this->app->make(Sandbox::class);
        $request = $this->prepareRequest($req);

        try {
            $sandbox->setRequest($request);

            $sandbox->init();

            $response = $sandbox->run($request);

            $this->sendResponse($sandbox, $response, $res);
        } catch (Throwable $e) {
            try {
                $exceptionResponse = $this->app
                    ->make(Handle::class)
                    ->render($request, $e);

                $this->sendResponse($sandbox, $exceptionResponse, $res);
            } catch (Throwable $e) {
                $this->logServerError($e);
            }
        } finally {
            $sandbox->clear();
        }
    }
900974
def __send_retry_requests(self, last_send_failure_time):
        """"""
        # make sure multiple failures having set multiple times do not run concurrently
        with self.__send_retry_requests_lock:
            with self.__requests:
                # produce list instead of generator as requests mapping can change during subsequent loop
                retry_reqs = [req for req in self.__requests.values()
                              if req._sent_without_response(last_send_failure_time)]

            retry_req_count = 0
            # don't continue if another network failure has occured (which will trigger this function again)
            while retry_reqs and self.__amqplink.last_send_exc_time <= last_send_failure_time:
                req = retry_reqs.pop()
                # lock individuallly so incoming request handling does not 'pause' for too long
                with self.__requests:
                    # might have received a response (or finished since)
                    if not (req.id_ in self.__requests and req._sent_without_response(last_send_failure_time)):
                        logger.debug('Not resending request %s (finished or has received response)', req.id_)
                        continue
                logger.debug('Resending request %s', req.id_)
                if not self.__retry_enqueue(PreparedMessage(req._inner_msg_out, req.id_)):
                    # client shutdown
                    break
                retry_req_count += 1

        if retry_req_count:
            logger.debug('Resending of %d request(s) complete (before %s)', retry_req_count, last_send_failure_time)
212989
public function process( array $reqparams, $inputstream )
	{
		if( isset( $reqparams['method'] ) ) {
			return $this->_processRequest( $reqparams );
		}

		return $this->_processStream( $inputstream );
	}
475385
protected function _connect(
        $host, $port, $timeout, $secure, $context, $retries = 0
    )
    {
        $conn = '';
        if (!strpos($host, '://')) {
            switch (strval($secure)) {
            case 'ssl':
            case 'sslv2':
            case 'sslv3':
                $conn = $secure . '://';
                $this->_secure = true;
                break;

            case 'tlsv1':
                $conn = 'tls://';
                $this->_secure = true;
                break;

            case 'tls':
            default:
                $conn = 'tcp://';
                break;
            }
        }
        $conn .= $host;
        if ($port) {
            $conn .= ':' . $port;
        }

        $this->_stream = @stream_socket_client(
            $conn,
            $error_number,
            $error_string,
            $timeout,
            STREAM_CLIENT_CONNECT,
            stream_context_create($context)
        );

        if ($this->_stream === false) {
            /* From stream_socket_client() page: a function return of false,
             * with an error code of 0, indicates a "problem initializing the
             * socket". These kind of issues are seen on the same server
             * (and even the same user account) as sucessful connections, so
             * these are likely transient issues. Retry up to 3 times in these
             * instances. */
            if (!$error_number && ($retries < 3)) {
                return $this->_connect($host, $port, $timeout, $secure, ++$retries, $context);
            }

            $e = new Client\Exception(
                'Error connecting to server.'
            );
            $e->details = sprintf("[%u] %s", $error_number, $error_string);
            throw $e;
        }

        stream_set_timeout($this->_stream, $timeout);

        if (function_exists('stream_set_read_buffer')) {
            stream_set_read_buffer($this->_stream, 0);
        }
        stream_set_write_buffer($this->_stream, 0);

        $this->_connected = true;
    }
872603
public function onReadUdp($stream = null, $events = 0, $arg = null)
    {
        if (Daemon::$process->reload) {
            return false;
        }

        if ($this->pool->maxConcurrency) {
            if ($this->pool->count() >= $this->pool->maxConcurrency) {
                $this->overload = true;
                return false;
            }
        }

        $host = null;
        do {
            $l = @socket_recvfrom($this->fd, $buf, 10240, MSG_DONTWAIT, $host, $port);
            if (!$l) {
                break;
            }
            $key = '[' . $host . ']:' . $port;
            if (!isset($this->portsMap[$key])) {
                if ($this->pool->allowedClients !== null) {
                    if (!self::netMatch($this->pool->allowedClients, $host)) {
                        Daemon::log('Connection is not allowed (' . $host . ')');
                    }
                    continue;
                }
                $class = $this->pool->connectionClass;
                $conn = new $class(null, $this->pool);
                $conn->setDgram(true);
                $conn->onWriteEv(null);
                $conn->setPeername($host, $port);
                $conn->setParentSocket($this);
                $this->portsMap[$key] = $conn;
                $conn->timeoutRef = setTimeout(function ($timer) use ($conn) {
                    $conn->finish();
                    $timer->finish();
                }, $conn->timeout * 1e6);
                $conn->onUdpPacket($buf);
            } else {
                $conn = $this->portsMap[$key];
                $conn->onUdpPacket($buf);
                Timer::setTimeout($conn->timeoutRef);
            }
        } while (true);
        return $host !== null;
    }
846636
public static function doExecute(Client $client, Request $req)
  {
    $httpRequest = $client->getIo()->makeRequest($req);
    $httpRequest->setExpectedClass($req->getExpectedClass());
    return self::decodeHttpResponse($httpRequest, $client);
  }
709346
public function requestOut($req, $s)
    {
        if ($this->write($s) === false) {
            $req->abort();
            return false;
        }
        return true;
    }
436598
public function tunnel(ClientSocket $socket, string $authority): Promise {
        return call(function () use ($socket, $authority) {
            $parser = new Parser(null);
            $parser->enqueueResponseMethodMatch("CONNECT");

            try {
                yield $socket->write("CONNECT {$authority} HTTP/1.1\r\n\r\n");
            } catch (StreamException $e) {
                new SocketException(
                    'Proxy CONNECT failed: Socket went away while writing tunneling request',
                    0,
                    $e
                );
            }

            try {
                while (null !== $chunk = yield $socket->read()) {
                    if (!$response = $parser->parse($chunk)) {
                        continue;
                    }

                    if ($response["status"] === 200) {
                        // Tunnel connected! We're finished \o/ #WinningAtLife #DealWithIt
                        \stream_context_set_option($socket->getResource(), 'artax*', 'is_tunneled', true);
                        return $socket->getResource();
                    }

                    throw new HttpException(\sprintf(
                        'Proxy CONNECT failed: Unexpected response status received from proxy: %d',
                        $response["status"]
                    ));
                }
            } catch (ParseException $e) {
                throw new HttpException(
                    'Proxy CONNECT failed: Malformed HTTP response received from proxy while establishing tunnel',
                    0,
                    $e
                );
            } catch (StreamException $e) {
                // fall through
            }

            throw new SocketException(
                'Proxy CONNECT failed: Socket went away while awaiting tunneling response',
                0,
                $e ?? null
            );
        });
    }
202861
protected function _initRequest($socket, $response)
    {
        $stream = $this->_classes['stream'];

        $curlOptions = $socket->options();
        $curlOptions[CURLOPT_INFILESIZE] = $socket->outgoing()->length();

        if (isset($curlOptions[CURLOPT_FILE])) {
            $response->body(fopen($curlOptions[CURLOPT_FILE], 'w+'));
        }
        $socket->incoming($response->stream());

        if (!$handle = curl_init()) {
            throw new ClientException('Unable to create a new cURL handle');
        }
        curl_setopt_array($handle, $curlOptions);
        return $handle;
    }