thrift-server.h

Go to the documentation of this file.

// Copyright 2012 Cloudera 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.


#ifndef IMPALA_RPC_THRIFT_SERVER_H
#define IMPALA_RPC_THRIFT_SERVER_H

#include <boost/thread/mutex.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/unordered_map.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <thrift/server/TServer.h>
#include <thrift/TProcessor.h>

#include "common/status.h"
#include "rpc/auth-provider.h"
#include "util/metrics.h"
#include "util/thread.h"

namespace impala {

class ThriftServer {
 public:
  typedef std::string Username;

  struct ConnectionContext {
    TUniqueId connection_id;
    Username username;
    TNetworkAddress network_address;
    std::string server_name;
  };

  class ConnectionHandlerIf {
   public:
    virtual void ConnectionStart(const ConnectionContext& connection_context) = 0;

    virtual void ConnectionEnd(const ConnectionContext& connection_context) = 0;
  };

  static const int DEFAULT_WORKER_THREADS = 2;

  enum ServerType { ThreadPool = 0, Threaded };

  ThriftServer(const std::string& name,
      const boost::shared_ptr<apache::thrift::TProcessor>& processor, int port,
      AuthProvider* auth_provider = NULL, MetricGroup* metrics = NULL,
      int num_worker_threads = DEFAULT_WORKER_THREADS, ServerType server_type = Threaded);

  Status EnableSsl(const std::string& certificate, const std::string& private_key);

  int port() const { return port_; }

  bool ssl_enabled() const { return ssl_enabled_; }

  void Join();

  void StopForTesting();

  Status Start();

  void SetConnectionHandler(ConnectionHandlerIf* connection) {
    connection_handler_ = connection;
  }

  static const TUniqueId& GetThreadConnectionId();

  static const ConnectionContext* GetThreadConnectionContext();

 private:
  Status CreateSocket(
      boost::shared_ptr<apache::thrift::transport::TServerTransport>* socket);

  bool started_;

  int port_;

  bool ssl_enabled_;

  std::string certificate_path_;

  std::string private_key_path_;

  int num_worker_threads_;

  ServerType server_type_;

  const std::string name_;

  boost::scoped_ptr<Thread> server_thread_;

  boost::scoped_ptr<apache::thrift::server::TServer> server_;
  boost::shared_ptr<apache::thrift::TProcessor> processor_;

  ConnectionHandlerIf* connection_handler_;

  boost::mutex connection_contexts_lock_;

  typedef boost::unordered_map<ConnectionContext*, boost::shared_ptr<ConnectionContext> >
      ConnectionContextSet;
  ConnectionContextSet connection_contexts_;

  bool metrics_enabled_;

  IntGauge* num_current_connections_metric_;

  IntCounter* total_connections_metric_;

  boost::uuids::random_generator uuid_generator_;

  AuthProvider* auth_provider_;

  class ThriftServerEventProcessor;
  friend class ThriftServerEventProcessor;
};

}

#endif