Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #ifndef config_amqp_client_h
       7             : #define config_amqp_client_h
       8             : 
       9             : #include <atomic>
      10             : #include <string>
      11             : #include <vector>
      12             : 
      13             : #include <boost/asio/ip/tcp.hpp>
      14             : 
      15             : #include <SimpleAmqpClient/SimpleAmqpClient.h>
      16             : 
      17             : struct ConfigClientOptions;
      18             : class ConfigClientManager;
      19             : struct ConfigAmqpConnInfo;
      20             : 
      21             : // Interface to AmqpClient
      22             : class ConfigAmqpChannel {
      23             : public:
      24           2 :     ConfigAmqpChannel() { }
      25           2 :     virtual ~ConfigAmqpChannel() { }
      26             : 
      27           0 :     virtual AmqpClient::Channel::ptr_t CreateFromUri(std::string uri) {
      28           0 :         return (channel_ = AmqpClient::Channel::CreateFromUri(uri));
      29             :     }
      30             : 
      31           0 :     virtual AmqpClient::Channel::ptr_t CreateSecure(
      32             :             std::string ca_cert, std::string host, std::string client_key,
      33             :             std::string client_cert, int port, std::string username,
      34             :             std::string password, std::string vhost, int frame_max = 131072,
      35             :             bool verify_hostname = false) {
      36           0 :         return (channel_ = AmqpClient::Channel::CreateSecure(ca_cert, host,
      37             :                     client_key, client_cert, port, username, password, vhost,
      38           0 :                     frame_max , verify_hostname));
      39             :     }
      40             : 
      41           0 :     virtual void DeclareExchange(const std::string &exchange_name,
      42             :          const std::string &exchange_type, bool passive, bool durable,
      43             :          bool auto_delete) {
      44           0 :         channel_->DeclareExchange(exchange_name, exchange_type, passive,
      45             :                                   durable, auto_delete);
      46           0 :     }
      47             : 
      48           0 :     virtual void DeleteQueue(const std::string &queue_name, bool if_unused,
      49             :                              bool if_empty) {
      50           0 :         channel_->DeleteQueue(queue_name, if_unused, if_empty);
      51           0 :     }
      52             : 
      53           0 :     virtual std::string DeclareQueue(const std::string &queue_name,
      54             :                 bool passive, bool durable, bool exclusive, bool auto_delete) {
      55             :         return channel_->DeclareQueue(queue_name, passive, durable, exclusive,
      56           0 :                                       auto_delete);
      57             :     }
      58             : 
      59           0 :     virtual void BindQueue(const std::string &queue_name,
      60             :                            const std::string &exchange_name,
      61             :                            const std::string &routing_key = "") {
      62           0 :         channel_->BindQueue(queue_name, exchange_name, routing_key);
      63           0 :     }
      64             : 
      65           0 :     virtual std::string BasicConsume(const std::string &queue,
      66             :                 const std::string &consumer_tag, bool no_local, bool no_ack,
      67             :                 bool exclusive, boost::uint16_t message_prefetch_count) {
      68             :         return channel_->BasicConsume(queue, consumer_tag, no_local, no_ack,
      69           0 :                                       exclusive, message_prefetch_count);
      70             :     }
      71             : 
      72           0 :     virtual bool BasicConsumeMessage(const std::string &consumer_tag,
      73             :          AmqpClient::Envelope::ptr_t &envelope, int timeout) {
      74           0 :         return channel_->BasicConsumeMessage(consumer_tag, envelope, timeout);
      75             :     }
      76             : 
      77           0 :     virtual void BasicAck(const AmqpClient::Envelope::ptr_t &message) {
      78           0 :         channel_->BasicAck(message);
      79           0 :     }
      80             : 
      81             : private:
      82             :     AmqpClient::Channel::ptr_t channel_;
      83             : };
      84             : 
      85             : /*
      86             :  * This is class interacts with RabbitMQ
      87             :  */
      88             : class ConfigAmqpClient {
      89             : public:
      90             :     typedef boost::asio::ip::tcp::endpoint Endpoint;
      91             :     ConfigAmqpClient(ConfigClientManager *mgr, std::string hostname,
      92             :                  std::string module_name, const ConfigClientOptions &options);
      93        7354 :     virtual ~ConfigAmqpClient() { }
      94             : 
      95             :     void StartRabbitMQReader();
      96             : 
      97         385 :     std::string rabbitmq_ip() const {
      98         385 :         if (current_server_index_ >= rabbitmq_ips_.size())
      99         177 :             return "";
     100         208 :         return rabbitmq_ips_[current_server_index_];
     101             :     }
     102             : 
     103         385 :     std::string rabbitmq_port() const {
     104         385 :         if (current_server_index_ >= rabbitmq_ips_.size())
     105         177 :             return "";
     106         208 :         return rabbitmq_ports_[current_server_index_];
     107             :     }
     108             : 
     109           0 :     size_t rabbitmq_server_list_len() const {
     110           0 :         return rabbitmq_ips_.size();
     111             :     }
     112             : 
     113           0 :     void increment_rabbitmq_server_index() {
     114           0 :         if (rabbitmq_ips_.size()) {
     115           0 :             current_server_index_ =
     116           0 :                 ((current_server_index_ + 1) % rabbitmq_ips_.size());
     117             :         }
     118           0 :     }
     119             : 
     120         104 :     std::string rabbitmq_user() const { return rabbitmq_user_; }
     121         104 :     std::string rabbitmq_password() const { return rabbitmq_password_; }
     122         495 :     std::string rabbitmq_vhost() const { return rabbitmq_vhost_; }
     123         104 :     bool rabbitmq_use_ssl() const { return rabbitmq_use_ssl_; }
     124             :     std::string rabbitmq_ssl_version() const { return rabbitmq_ssl_version_; }
     125           0 :     std::string rabbitmq_ssl_keyfile() const { return rabbitmq_ssl_keyfile_; }
     126           0 :     std::string rabbitmq_ssl_certfile() const { return rabbitmq_ssl_certfile_; }
     127           0 :     std::string rabbitmq_ssl_ca_certs() const { return rabbitmq_ssl_ca_certs_; }
     128             :     ConfigClientManager *config_manager() const { return mgr_; }
     129        1318 :     ConfigClientManager *config_manager() { return mgr_; }
     130         208 :     std::vector<Endpoint> endpoints() const { return endpoints_; }
     131           2 :     int reader_task_id() const { return reader_task_id_; }
     132         104 :     std::string hostname() const { return hostname_; }
     133         104 :     std::string module_name() const { return module_name_; }
     134             : 
     135          10 :     static void set_disable(bool disable) { disable_ = disable; }
     136             : 
     137             :     std::string FormAmqpUri(bool hide_auth_info) const;
     138             :     void EnqueueUUIDRequest(std::string oper, std::string obj_type,
     139             :                        std::string uuid_str);
     140             :     bool ProcessMessage(const std::string &json_message);
     141             :     void set_connected(bool connected);
     142             :     void GetConnectionInfo(ConfigAmqpConnInfo &info) const;
     143             :     // Test only
     144         206 :     bool terminate() const { return terminate_; }
     145           2 :     void set_terminate(bool terminate) { terminate_ = terminate; }
     146             : 
     147             : private:
     148             :     // A Job for reading the rabbitmq
     149             :     class RabbitMQReader;
     150             : 
     151             :     void ReportRabbitMQConnectionStatus(bool connected) const;
     152             : 
     153             :     ConfigClientManager *mgr_;
     154             :     std::string hostname_;
     155             :     std::string module_name_;
     156             : 
     157             :     int reader_task_id_;
     158             :     size_t current_server_index_;
     159             :     bool terminate_;
     160             :     std::vector<std::string> rabbitmq_ips_;
     161             :     std::vector<std::string> rabbitmq_ports_;
     162             :     std::string rabbitmq_user_;
     163             :     std::string rabbitmq_password_;
     164             :     std::string rabbitmq_vhost_;
     165             :     bool rabbitmq_use_ssl_;
     166             :     std::string rabbitmq_ssl_version_;
     167             :     std::string rabbitmq_ssl_keyfile_;
     168             :     std::string rabbitmq_ssl_certfile_;
     169             :     std::string rabbitmq_ssl_ca_certs_;
     170             :     static bool disable_;
     171             :     std::vector<Endpoint> endpoints_;
     172             :     std::atomic<bool> connection_status_;
     173             :     std::atomic<uint64_t> connection_status_change_at_;
     174             : };
     175             : 
     176             : #endif  // config_amqp_client_h