wangle::ObservingHandler< T, R, P > Class Template Reference

#include <ObservingHandler.h>

Inheritance diagram for wangle::ObservingHandler< T, R, P >:

Public Types
typedef HandlerAdapter< folly::IOBufQueue &, T >::Context	Context
Public Types inherited from wangle::HandlerAdapter< folly::IOBufQueue &, T >
typedef Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::Context	Context
Public Types inherited from wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >
typedef folly::IOBufQueue &	rin
typedef folly::IOBufQueue &	rout
typedef T	win
typedef T	wout
typedef HandlerContext< folly::IOBufQueue &, T >	Context

Public Member Functions
	ObservingHandler (const R &routingData, BroadcastPool< T, R, P > *broadcastPool)
	~ObservingHandler () override
	ObservingHandler (const ObservingHandler &)=delete
ObservingHandler &	operator= (const ObservingHandler &)=delete
	ObservingHandler (ObservingHandler &&)=default
ObservingHandler &	operator= (ObservingHandler &&)=default
void	transportActive (Context *ctx) override
void	readEOF (Context *ctx) override
void	readException (Context *ctx, folly::exception_wrapper ex) override
void	onNext (const T &buf) override
void	onError (folly::exception_wrapper ex) override
void	onCompleted () override
R &	routingData () override
Public Member Functions inherited from wangle::HandlerAdapter< folly::IOBufQueue &, T >
void	read (Context *ctx, folly::IOBufQueue &msg) override
folly::Future< folly::Unit >	write (Context *ctx, Tmsg) override
Public Member Functions inherited from wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >
	~Handler () override=default
virtual void	readEOF (Context *ctx)
virtual void	readException (Context *ctx, folly::exception_wrapper e)
virtual void	transportActive (Context *ctx)
virtual void	transportInactive (Context *ctx)
virtual folly::Future< folly::Unit >	writeException (Context *ctx, folly::exception_wrapper e)
virtual folly::Future< folly::Unit >	close (Context *ctx)
Public Member Functions inherited from wangle::HandlerBase< HandlerContext< folly::IOBufQueue &, T > >
virtual	~HandlerBase ()=default
virtual void	attachPipeline (HandlerContext< folly::IOBufQueue &, T > *)
virtual void	detachPipeline (HandlerContext< folly::IOBufQueue &, T > *)
HandlerContext< folly::IOBufQueue &, T > *	getContext ()
Public Member Functions inherited from wangle::Subscriber< T, R >
virtual	~Subscriber ()

Private Attributes
R	routingData_
BroadcastPool< T, R, P > *	broadcastPool_ {nullptr}
BroadcastHandler< T, R > *	broadcastHandler_ {nullptr}
uint64_t	subscriptionId_ {0}
bool	paused_ {false}
std::shared_ptr< bool >	deleted_ {new bool(false)}

Additional Inherited Members
Static Public Attributes inherited from wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >
static const HandlerDir	dir

Detailed Description

template<typename T, typename R, typename P = DefaultPipeline>
class wangle::ObservingHandler< T, R, P >

A Handler-Observer adaptor that can be used for subscribing to broadcasts. Maintains a thread-local BroadcastPool from which a BroadcastHandler is obtained and subscribed to based on the given routing data.

Definition at line 31 of file ObservingHandler.h.

Member Typedef Documentation

template<typename T, typename R, typename P = DefaultPipeline>

typedef HandlerAdapter<folly::IOBufQueue&, T>::Context wangle::ObservingHandler< T, R, P >::Context

Definition at line 34 of file ObservingHandler.h.

Constructor & Destructor Documentation

template<typename T, typename R, typename P>

wangle::ObservingHandler< T, R, P >::ObservingHandler	(	const R &	routingData,
		BroadcastPool< T, R, P > *	broadcastPool
	)

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

    : routingData_(routingData), broadcastPool_(CHECK_NOTNULL(broadcastPool)) {}

template<typename T , typename R , typename P >

wangle::ObservingHandler< T, R, P >::~ObservingHandler ( )

override

Definition at line 27 of file ObservingHandler-inl.h.

References wangle::ObservingHandler< T, R, P >::broadcastHandler_, wangle::ObservingHandler< T, R, P >::deleted_, and wangle::ObservingHandler< T, R, P >::subscriptionId_.

                                             {
  if (broadcastHandler_) {
    auto broadcastHandler = broadcastHandler_;
    broadcastHandler_ = nullptr;
    broadcastHandler->unsubscribe(subscriptionId_);
  }

  if (deleted_) {
    *deleted_ = true;
  }
}

template<typename T, typename R, typename P = DefaultPipeline>

wangle::ObservingHandler< T, R, P >::ObservingHandler ( const ObservingHandler< T, R, P > &  )

delete

template<typename T, typename R, typename P = DefaultPipeline>

wangle::ObservingHandler< T, R, P >::ObservingHandler ( ObservingHandler< T, R, P > &&  )

default

Member Function Documentation

template<typename T , typename R , typename P >

void wangle::ObservingHandler< T, R, P >::onCompleted ( )

overridevirtual

Implements wangle::Subscriber< T, R >.

Definition at line 115 of file ObservingHandler-inl.h.

References wangle::ObservingHandler< T, R, P >::broadcastHandler_, wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close(), and wangle::HandlerBase< HandlerContext< folly::IOBufQueue &, T > >::getContext().

                                            {
  // broadcastHandler_ will clear its subscribers and delete itself
  broadcastHandler_ = nullptr;
  this->close(this->getContext());
}

template<typename T , typename R , typename P >

void wangle::ObservingHandler< T, R, P >::onError ( folly::exception_wrapper  ex )

overridevirtual

Implements wangle::Subscriber< T, R >.

Definition at line 106 of file ObservingHandler-inl.h.

References wangle::ObservingHandler< T, R, P >::broadcastHandler_, wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close(), folly::exceptionStr(), and wangle::HandlerBase< HandlerContext< folly::IOBufQueue &, T > >::getContext().

Referenced by wangle::ObservingHandler< T, R, P >::transportActive().

                                                                 {
  LOG(ERROR) << "Error observing a broadcast: " << exceptionStr(ex);

  // broadcastHandler_ will clear its subscribers and delete itself
  broadcastHandler_ = nullptr;
  this->close(this->getContext());
}

template<typename T, typename R , typename P >

void wangle::ObservingHandler< T, R, P >::onNext ( const T &  buf )

overridevirtual

Implements wangle::Subscriber< T, R >.

Definition at line 91 of file ObservingHandler-inl.h.

References wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close(), wangle::ObservingHandler< T, R, P >::deleted_, wangle::HandlerBase< HandlerContext< folly::IOBufQueue &, T > >::getContext(), and wangle::HandlerAdapter< folly::IOBufQueue &, T >::write().

                                                    {
  auto ctx = this->getContext();
  auto deleted = deleted_;
  this->write(ctx, data)
      .onError([this, ctx, deleted](const std::exception& ex) {
        if (*deleted) {
          return;
        }

        LOG(ERROR) << "Error on write: " << ex.what();
        this->close(ctx);
      });
}

template<typename T, typename R, typename P = DefaultPipeline>

ObservingHandler& wangle::ObservingHandler< T, R, P >::operator= ( const ObservingHandler< T, R, P > &  )

delete

template<typename T, typename R, typename P = DefaultPipeline>

ObservingHandler& wangle::ObservingHandler< T, R, P >::operator= ( ObservingHandler< T, R, P > &&  )

default

template<typename T , typename R , typename P >

void wangle::ObservingHandler< T, R, P >::readEOF ( Context *  ctx )

override

Definition at line 78 of file ObservingHandler-inl.h.

References wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close().

                                                    {
  this->close(ctx);
}

template<typename T , typename R , typename P >

void wangle::ObservingHandler< T, R, P >::readException ( Context *  ctx, folly::exception_wrapper  ex  )

override

Definition at line 83 of file ObservingHandler-inl.h.

References wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close(), and folly::exceptionStr().

                               {
  LOG(ERROR) << "Error on read: " << exceptionStr(ex);
  this->close(ctx);
}

template<typename T , typename R , typename P >

R & wangle::ObservingHandler< T, R, P >::routingData ( )

overridevirtual

Implements wangle::Subscriber< T, R >.

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

References wangle::ObservingHandler< T, R, P >::routingData_.

Referenced by wangle::ObservingPipelineFactory< std::string, std::string >::newPipeline().

                                          {
  return routingData_;
}

template<typename T , typename R , typename P >

void wangle::ObservingHandler< T, R, P >::transportActive ( Context *  ctx )

override

Definition at line 40 of file ObservingHandler-inl.h.

References wangle::ObservingHandler< T, R, P >::broadcastHandler_, wangle::ObservingHandler< T, R, P >::broadcastPool_, wangle::Handler< folly::IOBufQueue &, folly::IOBufQueue &, T, T >::close(), wangle::ObservingHandler< T, R, P >::deleted_, wangle::ObservingHandler< T, R, P >::onError(), wangle::ObservingHandler< T, R, P >::routingData_, and wangle::ObservingHandler< T, R, P >::subscriptionId_.

                                                            {
  if (broadcastHandler_) {
    // Already connected
    return;
  }

  // Pause ingress until the remote connection is established and
  // broadcast handler is ready
  auto pipeline = dynamic_cast<ObservingPipeline<T>*>(ctx->getPipeline());
  CHECK(pipeline);
  pipeline->transportInactive();

  auto deleted = deleted_;
  broadcastPool_->getHandler(routingData_)
      .thenValue(
          [this, pipeline, deleted](BroadcastHandler<T, R>* broadcastHandler) {
            if (*deleted) {
              return;
            }

            broadcastHandler_ = broadcastHandler;
            subscriptionId_ = broadcastHandler_->subscribe(this);
            VLOG(10) << "Subscribed to a broadcast";

            // Resume ingress
            pipeline->transportActive();
          })
      .onError([this, ctx, deleted](const std::exception& ex) {
        if (*deleted) {
          return;
        }

        LOG(ERROR) << "Error subscribing to a broadcast: " << ex.what();
        this->close(ctx);
      });
}

Member Data Documentation

template<typename T, typename R, typename P = DefaultPipeline>

BroadcastHandler<T, R>* wangle::ObservingHandler< T, R, P >::broadcastHandler_ {nullptr}

private

Definition at line 62 of file ObservingHandler.h.

Referenced by wangle::ObservingHandler< T, R, P >::onCompleted(), wangle::ObservingHandler< T, R, P >::onError(), wangle::ObservingHandler< T, R, P >::transportActive(), and wangle::ObservingHandler< T, R, P >::~ObservingHandler().

template<typename T, typename R, typename P = DefaultPipeline>

BroadcastPool<T, R, P>* wangle::ObservingHandler< T, R, P >::broadcastPool_ {nullptr}

private

Definition at line 60 of file ObservingHandler.h.

Referenced by wangle::ObservingPipelineFactory< std::string, std::string >::broadcastPool(), and wangle::ObservingHandler< T, R, P >::transportActive().

template<typename T, typename R, typename P = DefaultPipeline>

std::shared_ptr<bool> wangle::ObservingHandler< T, R, P >::deleted_ {new bool(false)}

private

Definition at line 67 of file ObservingHandler.h.

Referenced by wangle::ObservingHandler< T, R, P >::onNext(), wangle::ObservingHandler< T, R, P >::transportActive(), and wangle::ObservingHandler< T, R, P >::~ObservingHandler().

template<typename T, typename R, typename P = DefaultPipeline>

bool wangle::ObservingHandler< T, R, P >::paused_ {false}

private

Definition at line 64 of file ObservingHandler.h.

template<typename T, typename R, typename P = DefaultPipeline>

R wangle::ObservingHandler< T, R, P >::routingData_

private

Definition at line 59 of file ObservingHandler.h.

Referenced by wangle::ObservingHandler< T, R, P >::routingData(), and wangle::ObservingHandler< T, R, P >::transportActive().

template<typename T, typename R, typename P = DefaultPipeline>

uint64_t wangle::ObservingHandler< T, R, P >::subscriptionId_ {0}

private

Definition at line 63 of file ObservingHandler.h.

Referenced by wangle::ObservingHandler< T, R, P >::transportActive(), and wangle::ObservingHandler< T, R, P >::~ObservingHandler().

---

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

• proxygen/wangle/wangle/channel/broadcast/ObservingHandler.h
• proxygen/wangle/wangle/channel/broadcast/ObservingHandler-inl.h