adamdebreceni commented on code in PR #1432:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1432#discussion_r1019075043
##########
extensions/mqtt/processors/AbstractMQTTProcessor.h:
##########
@@ -45,150 +43,166 @@ class AbstractMQTTProcessor : public core::Processor {
freeResources();
}
+ SMART_ENUM(MqttVersions,
+ (V_3X_AUTO, "3.x AUTO"),
+ (V_3_1_0, "3.1.0"),
+ (V_3_1_1, "3.1.1"),
+ (V_5_0, "5.0"));
+
+ SMART_ENUM(MqttQoS,
+ (LEVEL_0, "0"),
+ (LEVEL_1, "1"),
+ (LEVEL_2, "2"));
+
EXTENSIONAPI static const core::Property BrokerURI;
EXTENSIONAPI static const core::Property ClientID;
+ EXTENSIONAPI static const core::Property QoS;
+ EXTENSIONAPI static const core::Property MqttVersion;
+ EXTENSIONAPI static const core::Property ConnectionTimeout;
+ EXTENSIONAPI static const core::Property KeepAliveInterval;
+ EXTENSIONAPI static const core::Property LastWillTopic;
+ EXTENSIONAPI static const core::Property LastWillMessage;
+ EXTENSIONAPI static const core::Property LastWillQoS;
+ EXTENSIONAPI static const core::Property LastWillRetain;
+ EXTENSIONAPI static const core::Property LastWillContentType;
EXTENSIONAPI static const core::Property Username;
EXTENSIONAPI static const core::Property Password;
- EXTENSIONAPI static const core::Property KeepAliveInterval;
- EXTENSIONAPI static const core::Property MaxFlowSegSize;
- EXTENSIONAPI static const core::Property ConnectionTimeout;
- EXTENSIONAPI static const core::Property Topic;
- EXTENSIONAPI static const core::Property QoS;
EXTENSIONAPI static const core::Property SecurityProtocol;
EXTENSIONAPI static const core::Property SecurityCA;
EXTENSIONAPI static const core::Property SecurityCert;
EXTENSIONAPI static const core::Property SecurityPrivateKey;
EXTENSIONAPI static const core::Property SecurityPrivateKeyPassword;
- EXTENSIONAPI static const core::Property LastWillTopic;
- EXTENSIONAPI static const core::Property LastWillMessage;
- EXTENSIONAPI static const core::Property LastWillQoS;
- EXTENSIONAPI static const core::Property LastWillRetain;
- EXTENSIONAPI static auto properties() {
+
+ static auto basicProperties() {
+ return std::array{
+ BrokerURI,
+ ClientID,
+ MqttVersion
+ };
+ }
+
+ static auto advancedProperties() {
return std::array{
- BrokerURI,
- Topic,
- ClientID,
- QoS,
- ConnectionTimeout,
- KeepAliveInterval,
- MaxFlowSegSize,
- LastWillTopic,
- LastWillMessage,
- LastWillQoS,
- LastWillRetain,
- Username,
- Password,
- SecurityProtocol,
- SecurityCA,
- SecurityCert,
- SecurityPrivateKey,
- SecurityPrivateKeyPassword
+ QoS,
+ ConnectionTimeout,
+ KeepAliveInterval,
+ LastWillTopic,
+ LastWillMessage,
+ LastWillQoS,
+ LastWillRetain,
+ LastWillContentType,
+ Username,
+ Password,
+ SecurityProtocol,
+ SecurityCA,
+ SecurityCert,
+ SecurityPrivateKey,
+ SecurityPrivateKeyPassword
};
}
void onSchedule(const std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSessionFactory>& factory) override;
+ void onTrigger(const std::shared_ptr<core::ProcessContext>& context, const
std::shared_ptr<core::ProcessSession>& session) override;
void notifyStop() override {
freeResources();
}
protected:
+ /**
+ * Connect to MQTT broker. Synchronously waits until connection succeeds or
fails.
+ */
void reconnect();
+ /**
+ * Checks property consistency before connecting to broker
+ */
+ virtual void checkProperties() {
+ }
+
+ /**
+ * Checks broker limits and supported features vs our desired features after
connecting to broker
+ */
+ void checkBrokerLimits();
+ virtual void checkBrokerLimitsImpl() = 0;
+
+ // variables being used for a synchronous connection and disconnection
+ std::shared_mutex client_mutex_;
+
MQTTAsync client_ = nullptr;
std::string uri_;
- std::string topic_;
std::chrono::seconds keep_alive_interval_{60};
- uint64_t max_seg_size_ = std::numeric_limits<uint64_t>::max();
- std::chrono::seconds connection_timeout_{30};
- uint32_t qos_ = MQTT_QOS_1;
+ std::chrono::seconds connection_timeout_{10};
+ MqttQoS qos_{MqttQoS::LEVEL_0};
std::string clientID_;
std::string username_;
std::string password_;
+ MqttVersions mqtt_version_{MqttVersions::V_3X_AUTO};
- private:
- // MQTT async callback
- static int msgReceived(void *context, char* topic_name, int topic_len,
MQTTAsync_message* message) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onMessageReceived(topic_name, topic_len, message);
- return 1;
- }
+ // Supported operations
+ std::optional<bool> retain_available_;
+ std::optional<bool> wildcard_subscription_available_;
+ std::optional<bool> shared_subscription_available_;
- // MQTT async callback
- static void connectionLost(void *context, char* cause) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onConnectionLost(cause);
- }
+ std::optional<uint16_t> broker_topic_alias_maximum_;
+ std::optional<uint16_t> broker_receive_maximum_;
+ std::optional<uint8_t> maximum_qos_;
+ std::optional<uint32_t> maximum_packet_size_;
- // MQTT async callback
- static void connectionSuccess(void* context, MQTTAsync_successData*
response) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onConnectionSuccess(response);
- }
+ std::optional<std::chrono::seconds> maximum_session_expiry_interval_;
+ std::optional<std::chrono::seconds> server_keep_alive_;
- // MQTT async callback
- static void connectionFailure(void* context, MQTTAsync_failureData*
response) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onConnectionFailure(response);
- }
+ private:
+ /**
+ * Initializes local MQTT client and connects to broker.
+ */
+ void initializeClient();
- // MQTT async callback
- static void disconnectionSuccess(void* context, MQTTAsync_successData*
response) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onDisconnectionSuccess(response);
- }
+ /**
+ * Calls disconnect() and releases local MQTT client
+ */
+ void freeResources();
- // MQTT async callback
- static void disconnectionFailure(void* context, MQTTAsync_failureData*
response) {
- auto* processor = reinterpret_cast<AbstractMQTTProcessor*>(context);
- processor->onDisconnectionFailure(response);
- }
+ /**
+ * Disconnect from MQTT broker. Synchronously waits until disconnection
succeeds or fails.
+ */
+ void disconnect();
+
+ virtual void readProperties(const std::shared_ptr<core::ProcessContext>&
context) = 0;
+ virtual void onTriggerImpl(const std::shared_ptr<core::ProcessContext>&
context, const std::shared_ptr<core::ProcessSession>& session) = 0;
+ virtual void startupClient() = 0;
+ void setBrokerLimits(MQTTAsync_successData5* response);
+
+ // MQTT static async callbacks, calling their non-static counterparts with
context being pointer to "this"
+ static void connectionLost(void *context, char* cause);
+ static void connectionSuccess(void* context, MQTTAsync_successData*
response);
+ static void connectionSuccess5(void* context, MQTTAsync_successData5*
response);
+ static void connectionFailure(void* context, MQTTAsync_failureData*
response);
+ static void connectionFailure5(void* context, MQTTAsync_failureData5*
response);
+ static int msgReceived(void *context, char* topic_name, int topic_len,
MQTTAsync_message* message);
+
+ // MQTT async callback methods
+ void onConnectionLost(char* cause);
+ void onConnectFinished(MQTTAsync_successData* success_data,
MQTTAsync_successData5* success_data_5, MQTTAsync_failureData* failure_data,
MQTTAsync_failureData5* failure_data_5);
+ void onDisconnectFinished(MQTTAsync_successData* success_data,
MQTTAsync_successData5* success_data_5, MQTTAsync_failureData* failure_data,
MQTTAsync_failureData5* failure_data_5);
+ /**
+ * Called if message is received. This is default implementation, to be
overridden if subclass wants to use the message.
+ * @param topic_name name of the topic of message
+ * @param topic_len length of topic name
+ * @param message MQTT message
+ */
virtual void onMessageReceived(char* topic_name, int /*topic_len*/,
MQTTAsync_message* message) {
Review Comment:
I realize this is old code, but we could annotate these with
`gsl::owner<...>` or even create the appropriate `std::unique_ptr<...>`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
