Method Peek

be/src/rpc/TAcceptQueueServer.cpp:148–187

This function blocks until some bytes show up from the client. Returns true if some bytes are available from client; Returns false upon reading EOF, in which case the connection will be closed by the caller. If idle_poll_period_ms_ is not 0, this function will block up to idle_poll_period_ms_ milliseconds before waking up to check if the sessions associated with the connection have all expired du

  // due to inactivity. If so, it will return false and the connection
  // will be closed by the caller.
  bool Peek(void* connectionContext, TServerEventHandler* eventHandler) {
    // Set a timeout on input socket if idle_poll_period_ms_ is non-zero.
    TSocket* socket = static_cast<TSocket*>(transport_.get());
    if (server_.idle_poll_period_ms_ > 0) {
      socket->setRecvTimeout(server_.idle_poll_period_ms_);
    }

    // Block until some bytes show up or EOF or timeout.
    bool bytes_pending = true;
    for (;;) {
      try {
        bytes_pending = input_->getTransport()->peek();
        break;
      } catch (const TTransportException& ttx) {
        // Implementation of the underlying transport's peek() may call either
        // read() or peek() of the socket.
        if (eventHandler != nullptr && server_.idle_poll_period_ms_ > 0 &&
            (IsReadTimeoutTException(ttx) || IsPeekTimeoutTException(ttx))) {
          VLOG(2) << Substitute("Socket read or peek timeout encountered "
                                "(idle_poll_period_ms_=$0). $1",
              server_.idle_poll_period_ms_, ttx.what());
          ThriftServer::ThriftServerEventProcessor* thriftServerHandler =
              static_cast<ThriftServer::ThriftServerEventProcessor*>(eventHandler);
          if (thriftServerHandler->IsIdleContext(connectionContext)) {
            const string& client = socket->getSocketInfo();
            GlobalOutput.printf(
               "TAcceptQueueServer closing connection to idle client %s", client.c_str());
            bytes_pending = false;
            break;
          }
        } else {
          // Rethrow the exception to be handled by callers.
          throw;
        }
      }
    }
    // Unset the socket timeout.
    if (server_.idle_poll_period_ms_ > 0) socket->setRecvTimeout(0);
    return bytes_pending;
  }

  TAcceptQueueServer& server_;
  friend class TAcceptQueueServer;