proxygen: wangle::BroadcastHandler< T, R > Class Template Reference

#include <BroadcastHandler.h>

Inheritance diagram for wangle::BroadcastHandler< T, R >:

Public Types
typedef HandlerAdapter< T, std::unique_ptr< folly::IOBuf > >::Context	Context

Public Types inherited from wangle::HandlerAdapter< T, std::unique_ptr< folly::IOBuf > >
typedef Handler< T, T, std::unique_ptr< folly::IOBuf >, std::unique_ptr< folly::IOBuf > >::Context	Context

Public Types inherited from wangle::Handler< T, T, std::unique_ptr< folly::IOBuf >, std::unique_ptr< folly::IOBuf > >
typedef T	rin

typedef T	rout

typedef std::unique_ptr< folly::IOBuf >	win

typedef std::unique_ptr< folly::IOBuf >	wout

typedef HandlerContext< T, std::unique_ptr< folly::IOBuf > >	Context

Public Member Functions
	~BroadcastHandler () override

void	read (Context *ctx, T data) override

void	readEOF (Context *ctx) override

void	readException (Context *ctx, folly::exception_wrapper ex) override

virtual uint64_t	subscribe (Subscriber< T, R > *subscriber)

virtual void	unsubscribe (uint64_t subscriptionId)

virtual void	closeIfIdle ()

virtual void	onSubscribe (Subscriber< T, R > *)

virtual void	onUnsubscribe (Subscriber< T, R > *)

virtual void	onData (T &)

uint64_t	getArbitraryIdentifier ()

Public Member Functions inherited from wangle::HandlerAdapter< T, std::unique_ptr< folly::IOBuf > >
void	read (Context *ctx, Tmsg) override

folly::Future< folly::Unit >	write (Context *ctx, std::unique_ptr< folly::IOBuf >msg) override

Public Member Functions inherited from wangle::Handler< T, T, std::unique_ptr< folly::IOBuf >, std::unique_ptr< folly::IOBuf > >
	~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< T, std::unique_ptr< folly::IOBuf > > >
virtual	~HandlerBase ()=default

virtual void	attachPipeline (HandlerContext< T, std::unique_ptr< folly::IOBuf > > *)

virtual void	detachPipeline (HandlerContext< T, std::unique_ptr< folly::IOBuf > > *)

HandlerContext< T, std::unique_ptr< folly::IOBuf > > *	getContext ()

Protected Member Functions
template<typename FUNC >
void	forEachSubscriber (FUNC f)

Private Attributes
std::map< uint64_t, Subscriber< T, R > * >	subscribers_

uint64_t	nextSubscriptionId_ {0}

uint64_t	identifier_ {0}

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

Detailed Description

template<typename T, typename R>
class wangle::BroadcastHandler< T, R >

An Observable type handler for broadcasting/streaming data to a list of subscribers.

Definition at line 30 of file BroadcastHandler.h.

Member Typedef Documentation

template<typename T, typename R>

typedef HandlerAdapter<T, std::unique_ptr<folly::IOBuf> >::Context wangle::BroadcastHandler< T, R >::Context

Definition at line 32 of file BroadcastHandler.h.

Constructor & Destructor Documentation

template<typename T, typename R>

wangle::BroadcastHandler< T, R >::~BroadcastHandler ( )

inlineoverride

Definition at line 34 of file BroadcastHandler.h.

                                {
     CHECK(subscribers_.empty());
   }

Member Function Documentation

template<typename T , typename R >

void wangle::BroadcastHandler< T, R >::closeIfIdle ( )

virtual

If there are no subscribers listening to the broadcast, close the pipeline. This will also delete the broadcast from the BroadcastPool.

Definition at line 71 of file BroadcastHandler-inl.h.

References folly::netops::close().

Referenced by wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                          {
   if (subscribers_.empty()) {
     // No more subscribers. Clean up.
     // This will delete the broadcast from the pool.
     this->close(this->getContext());
   }
 }

template<typename T, typename R>

template<typename FUNC >

void wangle::BroadcastHandler< T, R >::forEachSubscriber ( FUNC f )

inlineprotected

Definition at line 87 of file BroadcastHandler.h.

                                  {
     auto subscribers = subscribers_;
     for (const auto& it : subscribers) {
       f(it.second);
     }
   }

template<typename T , typename R >

uint64_t wangle::BroadcastHandler< T, R >::getArbitraryIdentifier ( )

FOR TESTS ONLY! Return a unique identifier of this object.

Definition at line 80 of file BroadcastHandler-inl.h.

Referenced by wangle::BroadcastHandler< std::string, std::string >::onData(), and TEST_F().

                                                         {
   static std::atomic<uint64_t> identifierCounter{42};
   return identifier_ ? identifier_ : (identifier_ = ++identifierCounter);
 }

template<typename T, typename R>

virtual void wangle::BroadcastHandler< T, R >::onData ( T & )

inlinevirtual

Invoked for each data that is about to be broadcasted to the subscribers. Subclasses can override to add custom behavior.

Definition at line 77 of file BroadcastHandler.h.

77 {}

template<typename T, typename R>

virtual void wangle::BroadcastHandler< T, R >::onSubscribe ( Subscriber< T, R > * )

inlinevirtual

Invoked when a new subscriber is added. Subclasses can override to add custom behavior.

Definition at line 65 of file BroadcastHandler.h.

65 {}

template<typename T, typename R>

virtual void wangle::BroadcastHandler< T, R >::onUnsubscribe ( Subscriber< T, R > * )

inlinevirtual

Invoked when a subscriber is removed. Subclasses can override to add custom behavior.

Definition at line 71 of file BroadcastHandler.h.

71 {}

template<typename T, typename R >

void wangle::BroadcastHandler< T, R >::read	(	Context *	ctx,
		T	data
	)

override

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

References wangle::Subscriber< T, R >::onNext(), and s.

Referenced by wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                                   {
   onData(data);
   forEachSubscriber([&](Subscriber<T, R>* s) {
     s->onNext(data);
   });
 }

template<typename T , typename R >

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

override

Definition at line 29 of file BroadcastHandler-inl.h.

References wangle::Subscriber< T, R >::onCompleted(), and s.

Referenced by TEST_F(), and wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                              {
   forEachSubscriber([&](Subscriber<T, R>* s) {
     s->onCompleted();
   });
   subscribers_.clear();
   closeIfIdle();
 }

template<typename T , typename R >

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

override

Definition at line 38 of file BroadcastHandler-inl.h.

References folly::exceptionStr(), wangle::Subscriber< T, R >::onError(), and s.

Referenced by wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                                                    {
   LOG(ERROR) << "Error while reading from upstream for broadcast: "
              << exceptionStr(ex);
 
   forEachSubscriber([&](Subscriber<T, R>* s) {
     s->onError(ex);
   });
   subscribers_.clear();
   closeIfIdle();
 }

template<typename T, typename R>

uint64_t wangle::BroadcastHandler< T, R >::subscribe ( Subscriber< T, R > * subscriber )

virtual

Subscribes to the broadcast. Returns a unique subscription ID for this subscriber.

Definition at line 51 of file BroadcastHandler-inl.h.

Referenced by TEST_F(), and wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                                                        {
   auto subscriptionId = nextSubscriptionId_++;
   subscribers_[subscriptionId] = subscriber;
   onSubscribe(subscriber);
   return subscriptionId;
 }

template<typename T , typename R >

void wangle::BroadcastHandler< T, R >::unsubscribe ( uint64_t subscriptionId )

virtual

Unsubscribe from the broadcast. Closes the pipeline if the number of subscribers reaches zero.

Definition at line 59 of file BroadcastHandler-inl.h.

Referenced by wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

                                                                 {
   auto iter = subscribers_.find(subscriptionId);
   if (iter == subscribers_.end()) {
     return;
   }
 
   onUnsubscribe(iter->second);
   subscribers_.erase(iter);
   closeIfIdle();
 }

Member Data Documentation

template<typename T, typename R>

uint64_t wangle::BroadcastHandler< T, R >::identifier_ {0}

private

Definition at line 99 of file BroadcastHandler.h.

template<typename T, typename R>

uint64_t wangle::BroadcastHandler< T, R >::nextSubscriptionId_ {0}

private

Definition at line 96 of file BroadcastHandler.h.

template<typename T, typename R>

std::map<uint64_t, Subscriber<T, R>*> wangle::BroadcastHandler< T, R >::subscribers_

private

Definition at line 95 of file BroadcastHandler.h.

Referenced by wangle::BroadcastHandler< std::string, std::string >::forEachSubscriber(), and wangle::BroadcastHandler< std::string, std::string >::~BroadcastHandler().

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

proxygen/wangle/wangle/channel/broadcast/BroadcastHandler.h
proxygen/wangle/wangle/channel/broadcast/BroadcastHandler-inl.h

Public Types

Public Member Functions

Protected Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

template<typename T, typename R> class wangle::BroadcastHandler< T, R >

Member Typedef Documentation

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation

template<typename T, typename R>
class wangle::BroadcastHandler< T, R >