// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 KUDU_RPC_CONNECTION_H
#define KUDU_RPC_CONNECTION_H

#include <boost/intrusive/list.hpp>
#include <ev++.h>
#include <memory>
#include <stdint.h>
#include <unordered_map>

#include <limits>
#include <string>
#include <vector>

#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/rpc/outbound_call.h"
#include "kudu/rpc/sasl_client.h"
#include "kudu/rpc/sasl_server.h"
#include "kudu/rpc/inbound_call.h"
#include "kudu/rpc/transfer.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/net/socket.h"
#include "kudu/util/object_pool.h"
#include "kudu/util/status.h"

namespace kudu {
namespace rpc {

class DumpRunningRpcsRequestPB;
class RpcConnectionPB;
class ReactorThread;

//
// A connection between an endpoint and us.
//
// Inbound connections are created by AcceptorPools, which eventually schedule
// RegisterConnection() to be called from the reactor thread.
//
// Outbound connections are created by the Reactor thread in order to service
// outbound calls.
//
// Once a Connection is created, it can be used both for sending messages and
// receiving them, but any given connection is explicitly a client or server.
// If a pair of servers are making bidirectional RPCs, they will use two separate
// TCP connections (and Connection objects).
//
// This class is not fully thread-safe.  It is accessed only from the context of a
// single ReactorThread except where otherwise specified.
//
class Connection : public RefCountedThreadSafe<Connection> {
 public:
  enum Direction {
    // This host is sending calls via this connection.
    CLIENT,
    // This host is receiving calls via this connection.
    SERVER
  };

  // Create a new Connection.
  // reactor_thread: the reactor that owns us.
  // remote: the address of the remote end
  // socket: the socket to take ownership of.
  // direction: whether we are the client or server side
  Connection(ReactorThread *reactor_thread, Sockaddr remote, int socket,
             Direction direction);

  // Set underlying socket to non-blocking (or blocking) mode.
  Status SetNonBlocking(bool enabled);

  // Register our socket with an epoll loop.  We will only ever be registered in
  // one epoll loop at a time.
  void EpollRegister(ev::loop_ref& loop);

  ~Connection();

  MonoTime last_activity_time() const {
    return last_activity_time_;
  }

  // Returns true if we are not in the process of receiving or sending a
  // message, and we have no outstanding calls.
  bool Idle() const;

  // Fail any calls which are currently queued or awaiting response.
  // Prohibits any future calls (they will be failed immediately with this
  // same Status).
  void Shutdown(const Status &status);

  // Queue a new call to be made. If the queueing fails, the call will be
  // marked failed.
  // Takes ownership of the 'call' object regardless of whether it succeeds or fails.
  // This may be called from a non-reactor thread.
  void QueueOutboundCall(const std::shared_ptr<OutboundCall> &call);

  // Queue a call response back to the client on the server side.
  //
  // This may be called from a non-reactor thread.
  void QueueResponseForCall(gscoped_ptr<InboundCall> call);

  // The address of the remote end of the connection.
  const Sockaddr &remote() const { return remote_; }

  // Set the user credentials which should be used to log in.
  void set_user_credentials(const UserCredentials &user_credentials);

  // Modify the user credentials which will be used to log in.
  UserCredentials* mutable_user_credentials() { return &user_credentials_; }

  // Get the user credentials which will be used to log in.
  const UserCredentials &user_credentials() const { return user_credentials_; }

  // libev callback when data is available to read.
  void ReadHandler(ev::io &watcher, int revents);

  // libev callback when we may write to the socket.
  void WriteHandler(ev::io &watcher, int revents);

  // Safe to be called from other threads.
  std::string ToString() const;

  Direction direction() const { return direction_; }

  Socket *socket() { return &socket_; }

  // Return SASL client instance for this connection.
  SaslClient &sasl_client() { return sasl_client_; }

  // Return SASL server instance for this connection.
  SaslServer &sasl_server() { return sasl_server_; }

  // Initialize SASL client before negotiation begins.
  Status InitSaslClient();

  // Initialize SASL server before negotiation begins.
  Status InitSaslServer();

  // Go through the process of transferring control of the underlying socket back to the Reactor.
  void CompleteNegotiation(const Status &negotiation_status);

  // Indicate that negotiation is complete and that the Reactor is now in control of the socket.
  void MarkNegotiationComplete();

  Status DumpPB(const DumpRunningRpcsRequestPB& req,
                RpcConnectionPB* resp);

  ReactorThread *reactor_thread() const { return reactor_thread_; }

 private:
  friend struct CallAwaitingResponse;
  friend class QueueTransferTask;
  friend struct ResponseTransferCallbacks;

  // A call which has been fully sent to the server, which we're waiting for
  // the server to process. This is used on the client side only.
  struct CallAwaitingResponse {
    ~CallAwaitingResponse();

    // Notification from libev that the call has timed out.
    void HandleTimeout(ev::timer &watcher, int revents);

    Connection *conn;
    std::shared_ptr<OutboundCall> call;
    ev::timer timeout_timer;
  };

  typedef std::unordered_map<uint64_t, CallAwaitingResponse*> car_map_t;
  typedef std::unordered_map<uint64_t, InboundCall*> inbound_call_map_t;

  // Returns the next valid (positive) sequential call ID by incrementing a counter
  // and ensuring we roll over from INT32_MAX to 0.
  // Negative numbers are reserved for special purposes.
  int32_t GetNextCallId() {
    int32_t call_id = next_call_id_;
    if (PREDICT_FALSE(next_call_id_ == std::numeric_limits<int32_t>::max())) {
      next_call_id_ = 0;
    } else {
      next_call_id_++;
    }
    return call_id;
  }

  // An incoming packet has completed transferring on the server side.
  // This parses the call and delivers it into the call queue.
  void HandleIncomingCall(gscoped_ptr<InboundTransfer> transfer);

  // An incoming packet has completed on the client side. This parses the
  // call response, looks up the CallAwaitingResponse, and calls the
  // client callback.
  void HandleCallResponse(gscoped_ptr<InboundTransfer> transfer);

  // The given CallAwaitingResponse has elapsed its user-defined timeout.
  // Set it to Failed.
  void HandleOutboundCallTimeout(CallAwaitingResponse *car);

  // Queue a transfer for sending on this connection.
  // We will take ownership of the transfer.
  // This must be called from the reactor thread.
  void QueueOutbound(gscoped_ptr<OutboundTransfer> transfer);

  // The reactor thread that created this connection.
  ReactorThread * const reactor_thread_;

  // The socket we're communicating on.
  Socket socket_;

  // The remote address we're talking to.
  const Sockaddr remote_;

  // The credentials of the user operating on this connection (if a client user).
  UserCredentials user_credentials_;

  // whether we are client or server
  Direction direction_;

  // The last time we read or wrote from the socket.
  MonoTime last_activity_time_;

  // the inbound transfer, if any
  gscoped_ptr<InboundTransfer> inbound_;

  // notifies us when our socket is writable.
  ev::io write_io_;

  // notifies us when our socket is readable.
  ev::io read_io_;

  // Set to true when the connection is registered on a loop.
  // This is used for a sanity check in the destructor that we are properly
  // un-registered before shutting down.
  bool is_epoll_registered_;

  // waiting to be sent
  boost::intrusive::list<OutboundTransfer> outbound_transfers_; // NOLINT(*)

  // Calls which have been sent and are now waiting for a response.
  car_map_t awaiting_response_;

  // Calls which have been received on the server and are currently
  // being handled.
  inbound_call_map_t calls_being_handled_;

  // the next call ID to use
  int32_t next_call_id_;

  // Starts as Status::OK, gets set to a shutdown status upon Shutdown().
  Status shutdown_status_;

  // Temporary vector used when serializing - avoids an allocation
  // when serializing calls.
  std::vector<Slice> slices_tmp_;

  // Pool from which CallAwaitingResponse objects are allocated.
  // Also a funny name.
  ObjectPool<CallAwaitingResponse> car_pool_;
  typedef ObjectPool<CallAwaitingResponse>::scoped_ptr scoped_car;

  // SASL client instance used for connection negotiation when Direction == CLIENT.
  SaslClient sasl_client_;

  // SASL server instance used for connection negotiation when Direction == SERVER.
  SaslServer sasl_server_;

  // Whether we completed connection negotiation.
  bool negotiation_complete_;
};

} // namespace rpc
} // namespace kudu

#endif