Method onTrigger

extensions/mqtt/processors/ConvertUpdate.cpp:31–87

core::Property ConvertUpdate::SSLContext("SSL Context Service", "The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.", "");

void ConvertUpdate::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession>& /*session*/) {
  if (nullptr == mqtt_service_) {
    context->yield();
    return;
  }
  std::vector<uint8_t> update;
  bool received_update = false;
  while (mqtt_service_->get(100, listening_topic, update)) {
    // first we have the input topic string followed by the update URI
    if (update.size() > 0) {

      io::BufferStream stream(update.data(), update.size());

      std::string returnTopic, url;

      if (returnTopic.empty() || url.empty()) {
        logger_->log_debug("topic and/or URL are empty");
        break;
      }

      stream.read(returnTopic);
      stream.read(url);

      /**
       * Not having curl support is actually okay for MQTT to be built, but running the update processor requires
       * that we have curl available.
       */
      auto client_ptr = core::ClassLoader::getDefaultClassLoader().instantiateRaw("HTTPClient", "HTTPClient");
      if (nullptr == client_ptr) {
        logger_->log_error("Could not locate HTTPClient. You do not have cURL support!");
        return;
      }
      std::unique_ptr<utils::BaseHTTPClient> client = std::unique_ptr<utils::BaseHTTPClient>(dynamic_cast<utils::BaseHTTPClient*>(client_ptr));
      client->initialize("GET");
      client->setConnectionTimeout(std::chrono::milliseconds(2000));
      client->setReadTimeout(std::chrono::milliseconds(2000));

      if (client->submit()) {
        auto data = client->getResponseBody();
        std::vector<uint8_t> raw_data;
        std::transform(std::begin(data), std::end(data), std::back_inserter(raw_data), [](char c) {
          return (uint8_t)c;
        });
        mqtt_service_->send(returnTopic, raw_data);
      }

      received_update = true;
    } else {
      break;
    }
  }

  if (!received_update) {
    context->yield();
  }

}