| 54 | } |
| 55 | |
| 56 | void AbstractMQTTProcessor::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory>& /*factory*/) { |
| 57 | sslEnabled_ = false; |
| 58 | sslopts_ = MQTTClient_SSLOptions_initializer; |
| 59 | |
| 60 | std::string value; |
| 61 | int64_t valInt; |
| 62 | value = ""; |
| 63 | if (context->getProperty(BrokerURL.getName(), value) && !value.empty()) { |
| 64 | uri_ = value; |
| 65 | logger_->log_debug("AbstractMQTTProcessor: BrokerURL [%s]", uri_); |
| 66 | } |
| 67 | value = ""; |
| 68 | if (context->getProperty(ClientID.getName(), value) && !value.empty()) { |
| 69 | clientID_ = value; |
| 70 | logger_->log_debug("AbstractMQTTProcessor: ClientID [%s]", clientID_); |
| 71 | } |
| 72 | value = ""; |
| 73 | if (context->getProperty(Topic.getName(), value) && !value.empty()) { |
| 74 | topic_ = value; |
| 75 | logger_->log_debug("AbstractMQTTProcessor: Topic [%s]", topic_); |
| 76 | } |
| 77 | value = ""; |
| 78 | if (context->getProperty(UserName.getName(), value) && !value.empty()) { |
| 79 | userName_ = value; |
| 80 | logger_->log_debug("AbstractMQTTProcessor: UserName [%s]", userName_); |
| 81 | } |
| 82 | value = ""; |
| 83 | if (context->getProperty(PassWord.getName(), value) && !value.empty()) { |
| 84 | passWord_ = value; |
| 85 | logger_->log_debug("AbstractMQTTProcessor: PassWord [%s]", passWord_); |
| 86 | } |
| 87 | value = ""; |
| 88 | if (context->getProperty(CleanSession.getName(), value) && !value.empty() && |
| 89 | org::apache::nifi::minifi::utils::StringUtils::StringToBool(value, cleanSession_)) { |
| 90 | logger_->log_debug("AbstractMQTTProcessor: CleanSession [%d]", cleanSession_); |
| 91 | } |
| 92 | value = ""; |
| 93 | if (context->getProperty(KeepLiveInterval.getName(), value) && !value.empty()) { |
| 94 | core::TimeUnit unit; |
| 95 | if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) { |
| 96 | keepAliveInterval_ = valInt/1000; |
| 97 | logger_->log_debug("AbstractMQTTProcessor: KeepLiveInterval [%" PRId64 "]", keepAliveInterval_); |
| 98 | } |
| 99 | } |
| 100 | value = ""; |
| 101 | if (context->getProperty(ConnectionTimeOut.getName(), value) && !value.empty()) { |
| 102 | core::TimeUnit unit; |
| 103 | if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) { |
| 104 | connectionTimeOut_ = valInt/1000; |
| 105 | logger_->log_debug("AbstractMQTTProcessor: ConnectionTimeOut [%" PRId64 "]", connectionTimeOut_); |
| 106 | } |
| 107 | } |
| 108 | value = ""; |
| 109 | if (context->getProperty(QOS.getName(), value) && !value.empty() && (value == MQTT_QOS_0 || value == MQTT_QOS_1 || MQTT_QOS_2) && |
| 110 | core::Property::StringToInt(value, valInt)) { |
| 111 | qos_ = valInt; |
| 112 | logger_->log_debug("AbstractMQTTProcessor: QOS [%" PRId64 "]", qos_); |
| 113 | } |
nothing calls this directly
no test coverage detected