Line data    Source code

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