central_controller_rpc_client.h

Go to the documentation of this file.

// Copyright 2016 Google Inc.

#ifndef PAGESPEED_CONTROLLER_CENTRAL_CONTROLLER_RPC_CLIENT_H_
#define PAGESPEED_CONTROLLER_CENTRAL_CONTROLLER_RPC_CLIENT_H_

#include <memory>
#include <unordered_set>

#include "base/macros.h"
#include "pagespeed/controller/controller.grpc.pb.h"
#include "pagespeed/controller/central_controller.h"
#include "pagespeed/controller/expensive_operation_callback.h"
#include "pagespeed/controller/schedule_rewrite_callback.h"
#include "pagespeed/kernel/base/basictypes.h"
#include "pagespeed/kernel/util/grpc.h"
#include "pagespeed/kernel/base/abstract_mutex.h"
#include "pagespeed/kernel/base/message_handler.h"
#include "pagespeed/kernel/base/statistics.h"
#include "pagespeed/kernel/base/string.h"
#include "pagespeed/kernel/base/timer.h"
#include "pagespeed/kernel/base/thread_annotations.h"
#include "pagespeed/kernel/base/thread_system.h"

namespace net_instaweb {


class CentralControllerRpcClient : public CentralController {
 public:
  static const char kControllerReconnectTimeStatistic[];
  static const int kControllerReconnectDelayMs;

  CentralControllerRpcClient(const GoogleString& server_address,
                             int panic_threshold, ThreadSystem* thread_system,
                             Timer* timer, Statistics* statistics,
                             MessageHandler* handler);
  virtual ~CentralControllerRpcClient();

  void ScheduleExpensiveOperation(
      ExpensiveOperationCallback* callback) override;
  void ScheduleRewrite(ScheduleRewriteCallback* callback) override;

  void ShutDown() override LOCKS_EXCLUDED(mutex_);

  static void InitStats(Statistics* stats);

 private:
  enum State {
    DISCONNECTED,  
    RUNNING,
    SHUTDOWN,  
  };

  class GrpcClientThread;
  class ClientRegistry;

  template <typename ContextT, typename CallbackT>
  void StartContext(CallbackT* callback) LOCKS_EXCLUDED(mutex_);

  void ConsiderConnecting(int64 now_ms) EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  bool TimestampsAllowConnection(int64 now) EXCLUSIVE_LOCKS_REQUIRED(mutex_);

  ThreadSystem* thread_system_;
  Timer* timer_;
  std::unique_ptr<AbstractMutex> mutex_;
  std::unique_ptr<ClientRegistry> clients_;
  MessageHandler* handler_;

  State state_ GUARDED_BY(mutex_);
  const int controller_panic_threshold_;
  int64 reconnect_time_ms_ GUARDED_BY(mutex_);
  UpDownCounter* reconnect_time_ms_statistic_;

  std::unique_ptr<::grpc::CompletionQueue> queue_;
  std::shared_ptr<::grpc::ChannelInterface> channel_;
  std::unique_ptr<grpc::CentralControllerRpcService::Stub> stub_;

  std::unique_ptr<GrpcClientThread> client_thread_ GUARDED_BY(mutex_);

  
};

}  

#endif