extensions/mqtt/controllerservice/MQTTControllerService.h - nifi-minifi-cpp - Git at Google

 /**
  *
  * 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 LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_
 #define LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_

 #include <openssl/err.h>
 #include <openssl/ssl.h>
 #include <iostream>
 #include <memory>
 #include "core/Resource.h"
 #include "utils/StringUtils.h"
 #include "io/validation.h"
 #include "core/controller/ControllerService.h"
 #include "core/logging/LoggerConfiguration.h"
 #include "controllers/SSLContextService.h"
 #include "concurrentqueue.h"
 #include "MQTTClient.h"

 namespace org {
 namespace apache {
 namespace nifi {
 namespace minifi {
 namespace controllers {

 static constexpr const char* const MQTT_QOS_0 = "0";
 static constexpr const char* const MQTT_QOS_1 = "1";
 static constexpr const char* const MQTT_QOS_2 = "2";

 class Message {
  public:
   // empty constructor facilitates moves
   Message() {
   }
   explicit Message(const std::string &topic, void *data, size_t dataLen)
       : topic_(topic),
         data_((uint8_t*) data, ((uint8_t*)data + dataLen)) {
   }
   explicit Message(const Message &&other)
       : topic_(std::move(other.topic_)),
         data_(std::move(other.data_)) {
   }
   ~Message() {
   }

   Message &operator=(const Message &&other) {
     topic_ = std::move(other.topic_);
     data_ = std::move(other.data_);
     return *this;
   }
   std::string topic_;
   std::vector<uint8_t> data_;
 };

 /**
  * MQTTContextService provides a controller service for MQTT connectivity.
  *
  */
 class MQTTControllerService : public core::controller::ControllerService {
  public:
   explicit MQTTControllerService(const std::string &name, const std::string &id)
       : ControllerService(name, id),
         initialized_(false),
         client_(nullptr),
         keepAliveInterval_(0),
         connectionTimeOut_(0),
         qos_(2),
         ssl_context_service_(nullptr),
         logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
   }

   explicit MQTTControllerService(const std::string &name, utils::Identifier uuid = utils::Identifier())
       : ControllerService(name, uuid),
         initialized_(false),
         client_(nullptr),
         keepAliveInterval_(0),
         connectionTimeOut_(0),
         qos_(2),
         ssl_context_service_(nullptr),
         logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
   }

   explicit MQTTControllerService(const std::string &name, const std::shared_ptr<Configure> &configuration)
       : ControllerService(name),
         initialized_(false),
         client_(nullptr),
         keepAliveInterval_(0),
         connectionTimeOut_(0),
         qos_(2),
         ssl_context_service_(nullptr),
         logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
     setConfiguration(configuration);
     initialize();
   }

   static core::Property BrokerURL;
   static core::Property ClientID;
   static core::Property UserName;
   static core::Property Password;
   static core::Property CleanSession;
   static core::Property KeepLiveInterval;
   static core::Property ConnectionTimeOut;
   static core::Property Topic;
   static core::Property QOS;
   static core::Property SecurityProtocol;

   virtual void initialize();

   void yield() {

   }

   int send(const std::string &topic, const std::vector<uint8_t> &data) {
     int token;
     MQTTClient_message pubmsg = MQTTClient_message_initializer;
     const uint8_t *d = data.data();
     pubmsg.payload = const_cast<uint8_t*>(d);
     pubmsg.payloadlen = data.size();
     pubmsg.qos = qos_;
     pubmsg.retained = 0;

     auto resp = MQTTClient_publishMessage(client_, topic.c_str(), &pubmsg, &token);
     if (resp != MQTTCLIENT_SUCCESS) {
       return -1;
     }
     if (qos_ == 0) {
       std::unique_lock<std::mutex> lock(delivery_mutex_);
       delivered_[token] = true;
     }
     return token;
   }

   int send(const std::string &topic, const uint8_t *data, size_t dataSize) {
     int token;

     MQTTClient_message pubmsg = MQTTClient_message_initializer;
     pubmsg.payload = const_cast<uint8_t*>(data);
     pubmsg.payloadlen = dataSize;
     pubmsg.qos = qos_;
     pubmsg.retained = 0;

     auto resp = MQTTClient_publishMessage(client_, topic.c_str(), &pubmsg, &token);
     if (resp != MQTTCLIENT_SUCCESS) {
       return -1;
     }

     if (qos_ == 0) {
       std::unique_lock<std::mutex> lock(delivery_mutex_);
       delivered_[token] = true;
     }
     return token;
   }

   bool isRunning() {
     return getState() == core::controller::ControllerServiceState::ENABLED;
   }

   bool isWorkAvailable() {
     return false;
   }

   virtual void onEnable();

   void subscribeToTopic(const std::string newTopic) {
     std::lock_guard<std::mutex> lock(initialization_mutex_);
     if (topics_.find(newTopic) == topics_.end()) {
       MQTTClient_subscribe(client_, newTopic.c_str(), qos_);
       topics_[newTopic].size_approx();
     }
   }

   bool waitForDelivery(const uint64_t millisToWait, int token) {
     std::unique_lock<std::mutex> lock(delivery_mutex_);
     if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return delivered_[token] == true;})) {
       bool delivered = delivered_[token];
       delivered_.erase(token);
       return delivered;
     } else {
       delivered_.erase(token);
       return false;
     }
   }

   bool get(const uint64_t millisToWait, const std::string &topic, std::vector<uint8_t> &data) {
     std::unique_lock<std::mutex> lock(delivery_mutex_);
     if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return topics_[topic].size_approx() > 0;})) {
       Message resp;
       if (topics_[topic].try_dequeue(resp)) {
         data = std::move(resp.data_);
         return true;
       } else {
         return false;
       }
     } else {
       return false;
     }
   }

   bool awaitResponse(const uint64_t millisToWait, int token, const std::string &topic, std::vector<uint8_t> &data) {
     std::unique_lock<std::mutex> lock(delivery_mutex_);
     if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {
       return
       delivered_[token] == true;
     })) {
       bool delivered = delivered_[token];
       if (delivered) {
         if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return topics_[topic].size_approx() > 0;})) {
           Message resp;
           if (topics_[topic].try_dequeue(resp)) {
             data = std::move(resp.data_);
             return true;
           } else {
             return false;
           }
         } else {
           return false;
         }
       }
       delivered_.erase(token);
       return delivered;
     } else {
       delivered_.erase(token);
       return false;
     }
   }

  protected:

   void acknowledgeDelivery(MQTTClient_deliveryToken token) {
     std::lock_guard<std::mutex> lock(delivery_mutex_);
     // locked the mutex
     auto finder = delivered_.find(token);
     // only acknowledge delivery if we expect the delivery to occur, otherwise
     // we won't have any waiters.
     if (finder != delivered_.end()) {
       delivered_[token] = true;
     }
   }

   void enqueue(const std::string &topic, Message &&message) {
     std::unique_lock<std::mutex> lock(delivery_mutex_);
     topics_[topic].enqueue(std::move(message));
     delivery_notification_.notify_one();
   }

   static void deliveryCallback(void *context, MQTTClient_deliveryToken dt) {
     MQTTControllerService *service = (MQTTControllerService *) context;
     service->acknowledgeDelivery(dt);
   }

   static int receiveCallback(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
     MQTTControllerService *service = (MQTTControllerService *) context;
     std::string topic(topicName, topicLen == 0 ? strlen(topicName) : topicLen);
     Message queueMessage(topic, message->payload, message->payloadlen);
     service->enqueue(topic, std::move(queueMessage));
     MQTTClient_freeMessage(&message);
     MQTTClient_free(topicName);
     return 1;
   }
   static void reconnectCallback(void *context, char *cause) {
     MQTTControllerService *service = (MQTTControllerService *) context;
     service->reconnect();
   }

   bool reconnect() {
     if (!client_)
       return false;
     if (MQTTClient_isConnected(client_))
       return true;
     MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
     conn_opts.keepAliveInterval = keepAliveInterval_;
     conn_opts.cleansession = 1;
     if (!userName_.empty()) {
       conn_opts.username = userName_.c_str();
       conn_opts.password = passWord_.c_str();
     }
     if (ssl_context_service_ != nullptr)
       conn_opts.ssl = &sslopts_;
     if (MQTTClient_connect(client_, &conn_opts) != MQTTCLIENT_SUCCESS) {
       logger_->log_error("Failed to connect to MQTT broker %s", uri_);
       return false;
     }

     if (!topic_.empty()) {
       std::unique_lock<std::mutex> lock(delivery_mutex_);
       MQTTClient_subscribe(client_, topic_.c_str(), qos_);
     }
     return true;
   }

   virtual void initializeProperties();

   std::mutex initialization_mutex_;
   std::atomic<bool> initialized_;

   MQTTClient client_;
   std::string uri_;
   std::string topic_;
   int64_t keepAliveInterval_;
   int64_t connectionTimeOut_;
   int64_t qos_;
   std::string clientID_;
   std::string userName_;
   std::string passWord_;

  private:

   std::map<int, bool> delivered_;
   std::map<std::string, moodycamel::ConcurrentQueue<Message> > topics_;

   std::mutex delivery_mutex_;
   std::condition_variable delivery_notification_;

   MQTTClient_SSLOptions sslopts_;

   std::shared_ptr<controllers::SSLContextService> ssl_context_service_;

   std::shared_ptr<logging::Logger> logger_;

 };

 } /* namespace controllers */
 } /* namespace minifi */
 } /* namespace nifi */
 } /* namespace apache */
 } /* namespace org */

 #endif /* LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_ */
	/**
	*
	* 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 LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_
	#define LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_

	#include <openssl/err.h>
	#include <openssl/ssl.h>
	#include <iostream>
	#include <memory>
	#include "core/Resource.h"
	#include "utils/StringUtils.h"
	#include "io/validation.h"
	#include "core/controller/ControllerService.h"
	#include "core/logging/LoggerConfiguration.h"
	#include "controllers/SSLContextService.h"
	#include "concurrentqueue.h"
	#include "MQTTClient.h"

	namespace org {
	namespace apache {
	namespace nifi {
	namespace minifi {
	namespace controllers {

	static constexpr const char* const MQTT_QOS_0 = "0";
	static constexpr const char* const MQTT_QOS_1 = "1";
	static constexpr const char* const MQTT_QOS_2 = "2";

	class Message {
	public:
	// empty constructor facilitates moves
	Message() {
	}
	explicit Message(const std::string &topic, void *data, size_t dataLen)
	: topic_(topic),
	data_((uint8_t) data, ((uint8_t)data + dataLen)) {
	}
	explicit Message(const Message &&other)
	: topic_(std::move(other.topic_)),
	data_(std::move(other.data_)) {
	}
	~Message() {
	}

	Message &operator=(const Message &&other) {
	topic_ = std::move(other.topic_);
	data_ = std::move(other.data_);
	return *this;
	}
	std::string topic_;
	std::vector<uint8_t> data_;
	};

	/**
	* MQTTContextService provides a controller service for MQTT connectivity.
	*
	*/
	class MQTTControllerService : public core::controller::ControllerService {
	public:
	explicit MQTTControllerService(const std::string &name, const std::string &id)
	: ControllerService(name, id),
	initialized_(false),
	client_(nullptr),
	keepAliveInterval_(0),
	connectionTimeOut_(0),
	qos_(2),
	ssl_context_service_(nullptr),
	logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
	}

	explicit MQTTControllerService(const std::string &name, utils::Identifier uuid = utils::Identifier())
	: ControllerService(name, uuid),
	initialized_(false),
	client_(nullptr),
	keepAliveInterval_(0),
	connectionTimeOut_(0),
	qos_(2),
	ssl_context_service_(nullptr),
	logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
	}

	explicit MQTTControllerService(const std::string &name, const std::shared_ptr<Configure> &configuration)
	: ControllerService(name),
	initialized_(false),
	client_(nullptr),
	keepAliveInterval_(0),
	connectionTimeOut_(0),
	qos_(2),
	ssl_context_service_(nullptr),
	logger_(logging::LoggerFactory<MQTTControllerService>::getLogger()) {
	setConfiguration(configuration);
	initialize();
	}

	static core::Property BrokerURL;
	static core::Property ClientID;
	static core::Property UserName;
	static core::Property Password;
	static core::Property CleanSession;
	static core::Property KeepLiveInterval;
	static core::Property ConnectionTimeOut;
	static core::Property Topic;
	static core::Property QOS;
	static core::Property SecurityProtocol;

	virtual void initialize();

	void yield() {

	}

	int send(const std::string &topic, const std::vector<uint8_t> &data) {
	int token;
	MQTTClient_message pubmsg = MQTTClient_message_initializer;
	const uint8_t *d = data.data();
	pubmsg.payload = const_cast<uint8_t*>(d);
	pubmsg.payloadlen = data.size();
	pubmsg.qos = qos_;
	pubmsg.retained = 0;

	auto resp = MQTTClient_publishMessage(client_, topic.c_str(), &pubmsg, &token);
	if (resp != MQTTCLIENT_SUCCESS) {
	return -1;
	}
	if (qos_ == 0) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	delivered_[token] = true;
	}
	return token;
	}

	int send(const std::string &topic, const uint8_t *data, size_t dataSize) {
	int token;

	MQTTClient_message pubmsg = MQTTClient_message_initializer;
	pubmsg.payload = const_cast<uint8_t*>(data);
	pubmsg.payloadlen = dataSize;
	pubmsg.qos = qos_;
	pubmsg.retained = 0;

	auto resp = MQTTClient_publishMessage(client_, topic.c_str(), &pubmsg, &token);
	if (resp != MQTTCLIENT_SUCCESS) {
	return -1;
	}

	if (qos_ == 0) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	delivered_[token] = true;
	}
	return token;
	}

	bool isRunning() {
	return getState() == core::controller::ControllerServiceState::ENABLED;
	}

	bool isWorkAvailable() {
	return false;
	}

	virtual void onEnable();

	void subscribeToTopic(const std::string newTopic) {
	std::lock_guard<std::mutex> lock(initialization_mutex_);
	if (topics_.find(newTopic) == topics_.end()) {
	MQTTClient_subscribe(client_, newTopic.c_str(), qos_);
	topics_[newTopic].size_approx();
	}
	}

	bool waitForDelivery(const uint64_t millisToWait, int token) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return delivered_[token] == true;})) {
	bool delivered = delivered_[token];
	delivered_.erase(token);
	return delivered;
	} else {
	delivered_.erase(token);
	return false;
	}
	}

	bool get(const uint64_t millisToWait, const std::string &topic, std::vector<uint8_t> &data) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return topics_[topic].size_approx() > 0;})) {
	Message resp;
	if (topics_[topic].try_dequeue(resp)) {
	data = std::move(resp.data_);
	return true;
	} else {
	return false;
	}
	} else {
	return false;
	}
	}

	bool awaitResponse(const uint64_t millisToWait, int token, const std::string &topic, std::vector<uint8_t> &data) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {
	return
	delivered_[token] == true;
	})) {
	bool delivered = delivered_[token];
	if (delivered) {
	if (delivery_notification_.wait_for(lock, std::chrono::milliseconds(millisToWait), [&] {return topics_[topic].size_approx() > 0;})) {
	Message resp;
	if (topics_[topic].try_dequeue(resp)) {
	data = std::move(resp.data_);
	return true;
	} else {
	return false;
	}
	} else {
	return false;
	}
	}
	delivered_.erase(token);
	return delivered;
	} else {
	delivered_.erase(token);
	return false;
	}
	}

	protected:

	void acknowledgeDelivery(MQTTClient_deliveryToken token) {
	std::lock_guard<std::mutex> lock(delivery_mutex_);
	// locked the mutex
	auto finder = delivered_.find(token);
	// only acknowledge delivery if we expect the delivery to occur, otherwise
	// we won't have any waiters.
	if (finder != delivered_.end()) {
	delivered_[token] = true;
	}
	}

	void enqueue(const std::string &topic, Message &&message) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	topics_[topic].enqueue(std::move(message));
	delivery_notification_.notify_one();
	}

	static void deliveryCallback(void *context, MQTTClient_deliveryToken dt) {
	MQTTControllerService service = (MQTTControllerService ) context;
	service->acknowledgeDelivery(dt);
	}

	static int receiveCallback(void context, char topicName, int topicLen, MQTTClient_message *message) {
	MQTTControllerService service = (MQTTControllerService ) context;
	std::string topic(topicName, topicLen == 0 ? strlen(topicName) : topicLen);
	Message queueMessage(topic, message->payload, message->payloadlen);
	service->enqueue(topic, std::move(queueMessage));
	MQTTClient_freeMessage(&message);
	MQTTClient_free(topicName);
	return 1;
	}
	static void reconnectCallback(void context, char cause) {
	MQTTControllerService service = (MQTTControllerService ) context;
	service->reconnect();
	}

	bool reconnect() {
	if (!client_)
	return false;
	if (MQTTClient_isConnected(client_))
	return true;
	MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
	conn_opts.keepAliveInterval = keepAliveInterval_;
	conn_opts.cleansession = 1;
	if (!userName_.empty()) {
	conn_opts.username = userName_.c_str();
	conn_opts.password = passWord_.c_str();
	}
	if (ssl_context_service_ != nullptr)
	conn_opts.ssl = &sslopts_;
	if (MQTTClient_connect(client_, &conn_opts) != MQTTCLIENT_SUCCESS) {
	logger_->log_error("Failed to connect to MQTT broker %s", uri_);
	return false;
	}

	if (!topic_.empty()) {
	std::unique_lock<std::mutex> lock(delivery_mutex_);
	MQTTClient_subscribe(client_, topic_.c_str(), qos_);
	}
	return true;
	}

	virtual void initializeProperties();

	std::mutex initialization_mutex_;
	std::atomic<bool> initialized_;

	MQTTClient client_;
	std::string uri_;
	std::string topic_;
	int64_t keepAliveInterval_;
	int64_t connectionTimeOut_;
	int64_t qos_;
	std::string clientID_;
	std::string userName_;
	std::string passWord_;

	private:

	std::map<int, bool> delivered_;
	std::map<std::string, moodycamel::ConcurrentQueue<Message> > topics_;

	std::mutex delivery_mutex_;
	std::condition_variable delivery_notification_;

	MQTTClient_SSLOptions sslopts_;

	std::shared_ptr<controllers::SSLContextService> ssl_context_service_;

	std::shared_ptr<logging::Logger> logger_;

	};

	} /* namespace controllers */
	} /* namespace minifi */
	} /* namespace nifi */
	} /* namespace apache */
	} /* namespace org */

	#endif /* LIBMINIFI_INCLUDE_CONTROLLERS_MQTTCONTEXTSERVICE_H_ */