request_result_rpc_handler.h

Go to the documentation of this file.

// Copyright 2016 Google Inc.

#ifndef PAGESPEED_CONTROLLER_REQUEST_RESULT_RPC_HANDLER_H_
#define PAGESPEED_CONTROLLER_REQUEST_RESULT_RPC_HANDLER_H_

#include "base/logging.h"
#include "pagespeed/kernel/base/function.h"
#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/base/ref_counted_ptr.h"
#include "pagespeed/controller/rpc_handler.h"
#include "pagespeed/kernel/util/grpc.h"


namespace net_instaweb {

template <typename HandlerT, typename ControllerT, typename AsyncServiceT,
          typename RequestT, typename ResponseT>
class RequestResultRpcHandler
    : public RpcHandler<AsyncServiceT, RequestT, ResponseT> {
 public:
  typedef RefCountedPtr<RequestResultRpcHandler> RefPtr;

  virtual ~RequestResultRpcHandler() {}

  static void CreateAndStart(AsyncServiceT* service,
                             ::grpc::ServerCompletionQueue* cq,
                             ControllerT* controller) {
    (new HandlerT(service, cq, controller))->Start();
  }

 protected:
  RequestResultRpcHandler(AsyncServiceT* service,
                          ::grpc::ServerCompletionQueue* cq,
                          ControllerT* controller)
      : RpcHandler<AsyncServiceT, RequestT, ResponseT>::RpcHandler(service, cq),
        controller_(controller),
        state_(INIT) {}

  void Finish(const ::grpc::Status& status) {
    state_ = DONE;
    RpcHandler<AsyncServiceT, RequestT, ResponseT>::Finish(status);
  }

  ControllerT* controller() { return controller_; }

 private:
  enum State {
    INIT,
    WAITING_FOR_CONTROLLER,
    OPERATION_RUNNING,
    DONE,
  };

  class NotifyClientCallback : public Function {
   public:
    NotifyClientCallback(RequestResultRpcHandler* handler)
        : handler_(handler) {}

    void Run() override { handler_->NotifyClient(true ); }
    void Cancel() override { handler_->NotifyClient(false ); }

   private:
    RequestResultRpcHandler::RefPtr handler_;
  };

  virtual void HandleClientRequest(const RequestT& req, Function* cb) = 0;
  virtual void HandleClientResult(const RequestT& req) = 0;

  virtual void HandleOperationFailed() = 0;

  void NotifyClient(bool ok_to_rewrite);

  void HandleRequest(const RequestT& req) override;
  void HandleError() override;

  RpcHandler<AsyncServiceT, RequestT, ResponseT>* CreateHandler(
      AsyncServiceT* service, ::grpc::ServerCompletionQueue* cq) override {
    return new HandlerT(service, cq, controller());
  }

  void InitResponder(AsyncServiceT* service, ::grpc::ServerContext* ctx,
                     typename RpcHandler<AsyncServiceT, RequestT,
                                         ResponseT>::ReaderWriterT* responder,
                     ::grpc::ServerCompletionQueue* cq,
                     void* callback) override = 0;

  ControllerT* controller_;
  State state_;

  friend class NotifyClientCallback;
  friend class RequestResultRpcHandlerTest;

  
};

template <typename HandlerT, typename ControllerT, typename AsyncServiceT,
          typename RequestT, typename ResponseT>
void RequestResultRpcHandler<HandlerT, ControllerT, AsyncServiceT, RequestT,
                             ResponseT>::HandleRequest(const RequestT& req) {
  switch (state_) {
    case INIT:
      state_ = WAITING_FOR_CONTROLLER;
      HandleClientRequest(req, new NotifyClientCallback(this));
      break;

    case OPERATION_RUNNING:
      HandleClientResult(req);
      Finish(::grpc::Status());
      break;

    default:
      LOG(ERROR) << "HandleRequest in unexpected state: " << state_;
      Finish(::grpc::Status(::grpc::StatusCode::ABORTED,
                            "State machine error (HandleRequest)"));
  }
}

template <typename HandlerT, typename ControllerT, typename AsyncServiceT,
          typename RequestT, typename ResponseT>
void RequestResultRpcHandler<HandlerT, ControllerT, AsyncServiceT, RequestT,
                             ResponseT>::HandleError() {
  if (state_ == OPERATION_RUNNING) {
    HandleOperationFailed();
  }
  state_ = DONE;
}

template <typename HandlerT, typename ControllerT, typename AsyncServiceT,
          typename RequestT, typename ResponseT>
void RequestResultRpcHandler<HandlerT, ControllerT, AsyncServiceT, RequestT,
                             ResponseT>::NotifyClient(bool ok_to_proceed) {
  if (state_ != WAITING_FOR_CONTROLLER) {
    if (ok_to_proceed) {
      HandleOperationFailed();
    }
    if (state_ != DONE) {
      Finish(::grpc::Status(::grpc::StatusCode::ABORTED,
                            "State machine error (NotifyClient)"));
      LOG(DFATAL) << "NotifyClient in unexpected state: " << state_;
    }
    return;
  }

  ResponseT resp;
  resp.set_ok_to_proceed(ok_to_proceed);
  bool write_ok = this->Write(resp);
  if (write_ok && ok_to_proceed) {
    state_ = OPERATION_RUNNING;
  } else if (write_ok && !ok_to_proceed) {
    Finish(::grpc::Status());
  } else  {
    HandleOperationFailed();
    state_ = DONE;
  }
}

}  

#endif