wangle::AcceptRoutingHandler< Pipeline, R > Class Template Reference

#include <AcceptRoutingHandler.h>

Inheritance diagram for wangle::AcceptRoutingHandler< Pipeline, R >:

Public Member Functions
	AcceptRoutingHandler (ServerBootstrap< Pipeline > *server, std::shared_ptr< RoutingDataHandlerFactory< R >> routingHandlerFactory, std::shared_ptr< RoutingDataPipelineFactory< Pipeline, R >> childPipelineFactory)
void	read (Context *ctx, AcceptPipelineType conn) override
void	readEOF (Context *ctx) override
void	readException (Context *ctx, folly::exception_wrapper ex) override
void	onRoutingData (uint64_t connId, typename RoutingDataHandler< R >::RoutingData &routingData) override
void	onError (uint64_t connId, folly::exception_wrapper ex) override
size_t	getRoutingPipelineCount () const
Public Member Functions inherited from wangle::InboundHandler< AcceptPipelineType >
	~InboundHandler () override=default
virtual void	transportActive (Context *ctx)
virtual void	transportInactive (Context *ctx)
Public Member Functions inherited from wangle::HandlerBase< InboundHandlerContext< AcceptPipelineType > >
virtual	~HandlerBase ()=default
virtual void	attachPipeline (InboundHandlerContext< AcceptPipelineType > *)
virtual void	detachPipeline (InboundHandlerContext< AcceptPipelineType > *)
InboundHandlerContext< AcceptPipelineType > *	getContext ()
Public Member Functions inherited from wangle::RoutingDataHandler< R >::Callback
virtual	~Callback ()
virtual void	onRoutingData (uint64_t connId, RoutingData &routingData)=0

Private Member Functions
void	populateAcceptors ()
virtual DefaultPipeline::Ptr	newRoutingPipeline ()

Private Attributes
ServerBootstrap< Pipeline > *	server_
std::shared_ptr< RoutingDataHandlerFactory< R > >	routingHandlerFactory_
std::shared_ptr< RoutingDataPipelineFactory< Pipeline, R > >	childPipelineFactory_
std::vector< Acceptor * >	acceptors_
std::map< uint64_t, DefaultPipeline::Ptr >	routingPipelines_
uint64_t	nextConnId_ {0}

Additional Inherited Members
Public Types inherited from wangle::InboundHandler< AcceptPipelineType >
typedef AcceptPipelineType	rin
typedef AcceptPipelineType	rout
typedef folly::Unit	win
typedef folly::Unit	wout
typedef InboundHandlerContext< AcceptPipelineType >	Context
Static Public Attributes inherited from wangle::InboundHandler< AcceptPipelineType >
static const HandlerDir	dir

Detailed Description

template<typename Pipeline, typename R>
class wangle::AcceptRoutingHandler< Pipeline, R >

Definition at line 42 of file AcceptRoutingHandler.h.

Constructor & Destructor Documentation

template<typename Pipeline, typename R>

wangle::AcceptRoutingHandler< Pipeline, R >::AcceptRoutingHandler ( ServerBootstrap< Pipeline > *  server, std::shared_ptr< RoutingDataHandlerFactory< R >>  routingHandlerFactory, std::shared_ptr< RoutingDataPipelineFactory< Pipeline, R >>  childPipelineFactory  )

inline

Definition at line 45 of file AcceptRoutingHandler.h.

      : server_(CHECK_NOTNULL(server)),
        routingHandlerFactory_(routingHandlerFactory),
        childPipelineFactory_(childPipelineFactory) {}

Member Function Documentation

template<typename Pipeline, typename R>

size_t wangle::AcceptRoutingHandler< Pipeline, R >::getRoutingPipelineCount ( ) const

inline

Definition at line 65 of file AcceptRoutingHandler.h.

                                         {
    return routingPipelines_.size();
  }

template<typename Pipeline, typename R>

virtual DefaultPipeline::Ptr wangle::AcceptRoutingHandler< Pipeline, R >::newRoutingPipeline ( )

inlineprivatevirtual

Reimplemented in wangle::MockAcceptRoutingHandler.

Definition at line 71 of file AcceptRoutingHandler.h.

                                                  {
    return DefaultPipeline::create();
  }

template<typename Pipeline , typename R >

void wangle::AcceptRoutingHandler< Pipeline, R >::onError ( uint64_t  connId, folly::exception_wrapper  ex  )

overridevirtual

Implements wangle::RoutingDataHandler< R >::Callback.

Definition at line 122 of file AcceptRoutingHandler-inl.h.

References folly::exception_wrapper::what().

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::AcceptRoutingHandler().

                               {
  VLOG(4) << "Exception while parsing routing data: " << ex.what();

  // Notify all handlers of the exception
  auto ctx = getContext();
  auto pipeline =
      CHECK_NOTNULL(dynamic_cast<AcceptPipeline*>(ctx->getPipeline()));
  pipeline->readException(ex);

  // Delete the routing pipeline. This will close and delete the socket as well.
  routingPipelines_.erase(connId);
}

template<typename Pipeline , typename R>

void wangle::AcceptRoutingHandler< Pipeline, R >::onRoutingData ( uint64_t  connId, typename RoutingDataHandler< R >::RoutingData &  routingData  )

override

Definition at line 74 of file AcceptRoutingHandler-inl.h.

References wangle::RoutingDataHandler< R >::RoutingData::bufQueue, folly::gen::move, wangle::RoutingDataHandler< R >::RoutingData::routingData, folly::netops::socket(), and uint64_t.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::AcceptRoutingHandler().

                                                            {
  // Get the routing pipeline corresponding to this connection
  auto routingPipelineIter = routingPipelines_.find(connId);
  if (routingPipelineIter == routingPipelines_.end()) {
    VLOG(2) << "Connection has already been closed, "
               "or routed to a worker thread.";
    return;
  }
  auto routingPipeline = std::move(routingPipelineIter->second);
  routingPipelines_.erase(routingPipelineIter);

  // Fetch the socket from the pipeline and pause reading from the
  // socket
  auto socket = std::dynamic_pointer_cast<folly::AsyncSocket>(
      routingPipeline->getTransport());
  routingPipeline->transportInactive();
  socket->detachEventBase();

  // Hash based on routing data to pick a new acceptor
  uint64_t hash = std::hash<R>()(routingData.routingData);
  auto acceptor = acceptors_[hash % acceptors_.size()];

  // Switch to the new acceptor's thread
  acceptor->getEventBase()->runInEventBaseThread(
      [ =, routingData = std::move(routingData) ]() mutable {
        socket->attachEventBase(acceptor->getEventBase());

        auto routingHandler =
            routingPipeline->template getHandler<RoutingDataHandler<R>>();
        DCHECK(routingHandler);
        auto transportInfo = routingPipeline->getTransportInfo();
        auto pipeline = childPipelineFactory_->newPipeline(
            socket, routingData.routingData, routingHandler, transportInfo);

        auto connection =
            new typename ServerAcceptor<Pipeline>::ServerConnection(pipeline);
        acceptor->addConnection(connection);

        pipeline->transportActive();

        // Pass in the buffered bytes to the pipeline
        pipeline->read(routingData.bufQueue);
      });
}

template<typename Pipeline , typename R >

void wangle::AcceptRoutingHandler< Pipeline, R >::populateAcceptors ( )

private

Definition at line 138 of file AcceptRoutingHandler-inl.h.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::getRoutingPipelineCount().

                                                          {
  if (!acceptors_.empty()) {
    return;
  }
  CHECK(server_);
  server_->forEachWorker(
      [&](Acceptor* acceptor) { acceptors_.push_back(acceptor); });
}

template<typename Pipeline , typename R >

void wangle::AcceptRoutingHandler< Pipeline, R >::read ( Context *  ctx, AcceptPipelineType  conn  )

overridevirtual

Implements wangle::InboundHandler< AcceptPipelineType >.

Definition at line 21 of file AcceptRoutingHandler-inl.h.

References proxygen::localAddr, folly::gen::move, proxygen::peerAddr, folly::netops::socket(), and uint64_t.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::AcceptRoutingHandler().

                             {
  if (conn.type() != typeid(ConnInfo&)) {
    return;
  }

  populateAcceptors();

  const auto& connInfo = boost::get<ConnInfo&>(conn);
  auto socket = std::shared_ptr<folly::AsyncTransportWrapper>(
      connInfo.sock, folly::DelayedDestruction::Destructor());

  uint64_t connId = nextConnId_++;

  // Create a new routing pipeline for this connection to read from
  // the socket until it parses the routing data
  auto routingPipeline = newRoutingPipeline();
  routingPipeline->addBack(wangle::AsyncSocketHandler(socket));
  routingPipeline->addBack(routingHandlerFactory_->newHandler(connId, this));
  routingPipeline->finalize();

  // Initialize TransportInfo and set it on the routing pipeline
  auto transportInfo = std::make_shared<TransportInfo>(connInfo.tinfo);
  folly::SocketAddress localAddr, peerAddr;
  try {
    socket->getLocalAddress(&localAddr);
    socket->getPeerAddress(&peerAddr);
  } catch (...) {
    VLOG(2) << "Socket is no longer valid.";
    return;
  }
  transportInfo->localAddr = std::make_shared<folly::SocketAddress>(localAddr);
  transportInfo->remoteAddr = std::make_shared<folly::SocketAddress>(peerAddr);
  routingPipeline->setTransportInfo(transportInfo);

  routingPipeline->transportActive();
  routingPipelines_[connId] = std::move(routingPipeline);
}

template<typename Pipeline , typename R >

void wangle::AcceptRoutingHandler< Pipeline, R >::readEOF ( Context *  ctx )

overridevirtual

Reimplemented from wangle::InboundHandler< AcceptPipelineType >.

Definition at line 62 of file AcceptRoutingHandler-inl.h.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::AcceptRoutingHandler().

                                                        {
  // Null implementation to terminate the call in this handler
}

template<typename Pipeline , typename R >

void wangle::AcceptRoutingHandler< Pipeline, R >::readException ( Context *  ctx, folly::exception_wrapper  ex  )

overridevirtual

Reimplemented from wangle::InboundHandler< AcceptPipelineType >.

Definition at line 67 of file AcceptRoutingHandler-inl.h.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::AcceptRoutingHandler().

                            {
  // Null implementation to terminate the call in this handler
}

Member Data Documentation

template<typename Pipeline, typename R>

std::vector<Acceptor*> wangle::AcceptRoutingHandler< Pipeline, R >::acceptors_

private

Definition at line 80 of file AcceptRoutingHandler.h.

template<typename Pipeline, typename R>

std::shared_ptr<RoutingDataPipelineFactory<Pipeline, R> > wangle::AcceptRoutingHandler< Pipeline, R >::childPipelineFactory_

private

Definition at line 78 of file AcceptRoutingHandler.h.

Referenced by wangle::AcceptRoutingPipelineFactory< Pipeline, R >::newPipeline().

template<typename Pipeline, typename R>

uint64_t wangle::AcceptRoutingHandler< Pipeline, R >::nextConnId_ {0}

private

Definition at line 82 of file AcceptRoutingHandler.h.

template<typename Pipeline, typename R>

std::shared_ptr<RoutingDataHandlerFactory<R> > wangle::AcceptRoutingHandler< Pipeline, R >::routingHandlerFactory_

private

Definition at line 76 of file AcceptRoutingHandler.h.

Referenced by wangle::AcceptRoutingPipelineFactory< Pipeline, R >::newPipeline().

template<typename Pipeline, typename R>

std::map<uint64_t, DefaultPipeline::Ptr> wangle::AcceptRoutingHandler< Pipeline, R >::routingPipelines_

private

Definition at line 81 of file AcceptRoutingHandler.h.

Referenced by wangle::AcceptRoutingHandler< DefaultPipeline, char >::getRoutingPipelineCount().

template<typename Pipeline, typename R>

ServerBootstrap<Pipeline>* wangle::AcceptRoutingHandler< Pipeline, R >::server_

private

Definition at line 75 of file AcceptRoutingHandler.h.

Referenced by wangle::AcceptRoutingPipelineFactory< Pipeline, R >::newPipeline().

---

The documentation for this class was generated from the following files:

• proxygen/wangle/wangle/bootstrap/AcceptRoutingHandler.h
• proxygen/wangle/wangle/bootstrap/AcceptRoutingHandler-inl.h