BroadcastHandler.h

Go to the documentation of this file.

/*
 * Copyright 2017-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#pragma once

#include <wangle/channel/AsyncSocketHandler.h>
#include <wangle/channel/Handler.h>
#include <wangle/channel/Pipeline.h>
#include <wangle/channel/broadcast/Subscriber.h>

namespace wangle {

template <typename T, typename R>
class BroadcastHandler : public HandlerAdapter<T, std::unique_ptr<folly::IOBuf>> {
 public:
  typedef typename HandlerAdapter<T, std::unique_ptr<folly::IOBuf>>::Context Context;

  ~BroadcastHandler() override {
    CHECK(subscribers_.empty());
  }

  // BytesToBytesHandler implementation
  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& /* data */) {}

  uint64_t getArbitraryIdentifier();

 protected:
  template <typename FUNC> // FUNC: Subscriber<T, R>* -> void
  void forEachSubscriber(FUNC f) {
    auto subscribers = subscribers_;
    for (const auto& it : subscribers) {
      f(it.second);
    }
  }

 private:
  std::map<uint64_t, Subscriber<T, R>*> subscribers_;
  uint64_t nextSubscriptionId_{0};

  // For unit tests only.
  uint64_t identifier_{0};
};

template <typename T, typename R>
class BroadcastPipelineFactory
    : public PipelineFactory<DefaultPipeline> {
 public:
  DefaultPipeline::Ptr newPipeline(
      std::shared_ptr<folly::AsyncTransportWrapper> socket) override = 0;

  virtual BroadcastHandler<T, R>* getBroadcastHandler(
      DefaultPipeline* pipeline) noexcept = 0;

  virtual void setRoutingData(DefaultPipeline* pipeline,
                              const R& routingData) = 0;
};

} // namespace wangle

#include <wangle/channel/broadcast/BroadcastHandler-inl.h>