rpc_handler.h

Go to the documentation of this file.

#ifndef PAGESPEED_CONTROLLER_RPC_HANDLER_H_
#define PAGESPEED_CONTROLLER_RPC_HANDLER_H_

// Copyright 2016 Google Inc.

#include "base/logging.h"
#include "base/macros.h"
#include "pagespeed/kernel/base/function.h"
#include "pagespeed/kernel/base/ref_counted_ptr.h"
#include "pagespeed/kernel/util/grpc.h"

namespace net_instaweb {


template <typename AsyncService, typename RequestT, typename ResponseT>
class RpcHandler
    : public RefCounted<RpcHandler<AsyncService, RequestT, ResponseT>> {
 public:
  virtual ~RpcHandler() { }

 protected:
  typedef ::grpc::ServerAsyncReaderWriter<ResponseT, RequestT> ReaderWriterT;

  RpcHandler(AsyncService* service, ::grpc::ServerCompletionQueue* cq);

  bool Write(const ResponseT& resp);

  bool Finish(const ::grpc::Status& status);

  void Start();

 private:
  enum State {
    INIT,
    WAITING_FOR_FIRST_READ,
    RUNNING,
    FINISHED,
  };

  typedef RefCountedPtr<RpcHandler<AsyncService, RequestT, ResponseT>>
      RefPtrT;

  virtual void HandleRequest(const RequestT& req) = 0;

  virtual void HandleError() = 0;

  virtual void HandleWriteDone() { }

  virtual void InitResponder(AsyncService* service, ::grpc::ServerContext* ctx,
                             ReaderWriterT* responder,
                             ::grpc::ServerCompletionQueue* cq,
                             void* tag) = 0;

  virtual RpcHandler* CreateHandler(AsyncService* service,
                                    ::grpc::ServerCompletionQueue* cq) = 0;

  void InitDone(RefPtrT r);
  void AttemptRead(RefPtrT r);
  void ReadDone(RefPtrT r);
  void FinishDone(RefPtrT r);
  void WriteDone(RefPtrT r);
  void CallHandleError(RefPtrT r);

  bool IsClientWriteable() const;

  AsyncService* service_;
  ::grpc::ServerCompletionQueue* cq_;
  ::grpc::ServerContext ctx_;
  ReaderWriterT responder_;

  RequestT request_;

  State state_;
  bool write_outstanding_;

  
};

template <typename AsyncService, typename RequestT, typename ResponseT>
RpcHandler<AsyncService, RequestT, ResponseT>::RpcHandler(
    AsyncService* service, ::grpc::ServerCompletionQueue* cq)
    : service_(service),
      cq_(cq),
      responder_(&ctx_),
      state_(INIT),
      write_outstanding_(false) {
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::Start() {
  InitResponder(service_, &ctx_, &responder_, cq_,
                MakeFunction(this, &RpcHandler::InitDone,
                             &RpcHandler::CallHandleError, RefPtrT(this)));
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::InitDone(RefPtrT ref) {
  CreateHandler(service_, cq_)->Start(); 

  if (state_ != FINISHED) {
    state_ = WAITING_FOR_FIRST_READ;
    AttemptRead(ref);
  } else {
  }
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::AttemptRead(RefPtrT ref) {
  if (state_ != WAITING_FOR_FIRST_READ) {
    DCHECK_EQ(state_, RUNNING);
  }

  if (state_ != FINISHED) {
    responder_.Read(&request_, MakeFunction(this, &RpcHandler::ReadDone,
                                            &RpcHandler::CallHandleError, ref));
  }
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::ReadDone(RefPtrT ref) {
  if (state_ == WAITING_FOR_FIRST_READ) {
    state_ = RUNNING;
  }

  HandleRequest(request_);
  request_.Clear(); 

  if (state_ != FINISHED) {
    AttemptRead(ref);
  }
}

template <typename AsyncService, typename RequestT, typename ResponseT>
bool RpcHandler<AsyncService, RequestT, ResponseT>::Finish(
    const ::grpc::Status& status) {
  if (state_ == FINISHED) {
    return false;
  }
  if (IsClientWriteable()) {
    responder_.Finish(
        status, MakeFunction(this, &RpcHandler::FinishDone,
                             &RpcHandler::CallHandleError, RefPtrT(this)));
  }
  state_ = FINISHED;
  return true;
}

template <typename AsyncService, typename RequestT, typename ResponseT>
bool RpcHandler<AsyncService, RequestT, ResponseT>::Write(
    const ResponseT& response) {
  if (IsClientWriteable() && !write_outstanding_) {
    write_outstanding_ = true;
    responder_.Write(response,
                     MakeFunction(this, &RpcHandler::WriteDone,
                                  &RpcHandler::CallHandleError, RefPtrT(this)));
    return true;
  } else {
    return false;
  }
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::WriteDone(RefPtrT ref) {
  write_outstanding_ = false;
  HandleWriteDone();
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::FinishDone(RefPtrT ref) {
  DCHECK_EQ(state_, FINISHED);
}

template <typename AsyncService, typename RequestT, typename ResponseT>
void RpcHandler<AsyncService, RequestT, ResponseT>::CallHandleError(
    RefPtrT ref) {
  if (state_ == RUNNING) {
    HandleError();
  }
  state_ = FINISHED;
}

template <typename AsyncService, typename RequestT, typename ResponseT>
bool RpcHandler<AsyncService, RequestT, ResponseT>::IsClientWriteable() const {
  return state_ == WAITING_FOR_FIRST_READ || state_ == RUNNING;
}

}  

#endif