<commit_before>//
// Copyright Copyright 2009-2022, AMT – The Association For Manufacturing Technology (“AMT”)
// All rights reserved.
//
//    Licensed 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.
//

#include "mqtt_adapter.hpp"

#include <boost/log/trivial.hpp>
#include <boost/uuid/name_generator_sha1.hpp>

#include <mqtt/async_client.hpp>
#include <mqtt/setup_log.hpp>

#include "configuration/config_options.hpp"
#include "device_model/device.hpp"
#include "mqtt/mqtt_client_impl.hpp"
#include "pipeline/convert_sample.hpp"
#include "pipeline/deliver.hpp"
#include "pipeline/delta_filter.hpp"
#include "pipeline/duplicate_filter.hpp"
#include "pipeline/message_mapper.hpp"
#include "pipeline/period_filter.hpp"
#include "pipeline/shdr_token_mapper.hpp"
#include "pipeline/shdr_tokenizer.hpp"
#include "pipeline/timestamp_extractor.hpp"
#include "pipeline/topic_mapper.hpp"
#include "pipeline/upcase_value.hpp"

using namespace std;
namespace asio = boost::asio;

namespace mtconnect {
  using namespace observation;
  using namespace entity;
  using namespace pipeline;
  using namespace source::adapter;

  namespace source::adapter::mqtt_adapter {

    MqttAdapter::MqttAdapter(boost::asio::io_context &io,
                             pipeline::PipelineContextPtr pipelineContext,
                             const ConfigOptions &options, const boost::property_tree::ptree &block)
      : Adapter("MQTT", io, options),
        m_ioContext(io),
        m_strand(Source::m_strand),
        m_pipeline(pipelineContext, Source::m_strand)
    {
      GetOptions(block, m_options, options);
      AddOptions(block, m_options,
                 {{configuration::UUID, string()},
                  {configuration::Manufacturer, string()},
                  {configuration::Station, string()},
                  {configuration::Url, string()},
                  {configuration::MqttCaCert, string()}});

      AddDefaultedOptions(block, m_options,
                          {{configuration::Host, "localhost"s},
                           {configuration::Port, 1883},
                           {configuration::MqttTls, false},
                           {configuration::AutoAvailable, false},
                           {configuration::RealTime, false},
                           {configuration::RelativeTime, false}});
      loadTopics(block, m_options);

      m_handler = m_pipeline.makeHandler();
      auto clientHandler = make_unique<ClientHandler>();

      clientHandler->m_connecting = [this](shared_ptr<MqttClient> client) {
        m_handler->m_connecting(client->getIdentity());
      };

      clientHandler->m_connected = [this](shared_ptr<MqttClient> client) {
        m_handler->m_connected(client->getIdentity());
      };

      clientHandler->m_disconnected = [this](shared_ptr<MqttClient> client) {
        m_handler->m_disconnected(client->getIdentity());
      };

      clientHandler->m_receive = [this](shared_ptr<MqttClient> client, const std::string &topic,
                                        const std::string &payload) {
        m_handler->m_processMessage(topic, payload, client->getIdentity());
      };
<commit_msg>Fixed mqtt adapter host and port, use common MqttHost and port<commit_after>//
// Copyright Copyright 2009-2022, AMT – The Association For Manufacturing Technology (“AMT”)
// All rights reserved.
//
//    Licensed 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.
//

#include "mqtt_adapter.hpp"

#include <boost/log/trivial.hpp>
#include <boost/uuid/name_generator_sha1.hpp>

#include <mqtt/async_client.hpp>
#include <mqtt/setup_log.hpp>

#include "configuration/config_options.hpp"
#include "device_model/device.hpp"
#include "mqtt/mqtt_client_impl.hpp"
#include "pipeline/convert_sample.hpp"
#include "pipeline/deliver.hpp"
#include "pipeline/delta_filter.hpp"
#include "pipeline/duplicate_filter.hpp"
#include "pipeline/message_mapper.hpp"
#include "pipeline/period_filter.hpp"
#include "pipeline/shdr_token_mapper.hpp"
#include "pipeline/shdr_tokenizer.hpp"
#include "pipeline/timestamp_extractor.hpp"
#include "pipeline/topic_mapper.hpp"
#include "pipeline/upcase_value.hpp"

using namespace std;
namespace asio = boost::asio;

namespace mtconnect {
  using namespace observation;
  using namespace entity;
  using namespace pipeline;
  using namespace source::adapter;

  namespace source::adapter::mqtt_adapter {

    MqttAdapter::MqttAdapter(boost::asio::io_context &io,
                             pipeline::PipelineContextPtr pipelineContext,
                             const ConfigOptions &options, const boost::property_tree::ptree &block)
      : Adapter("MQTT", io, options),
        m_ioContext(io),
        m_strand(Source::m_strand),
        m_pipeline(pipelineContext, Source::m_strand)
    {
      GetOptions(block, m_options, options);
      AddOptions(block, m_options,
                 {{configuration::UUID, string()},
                  {configuration::Manufacturer, string()},
                  {configuration::Station, string()},
                  {configuration::Url, string()},
                  {configuration::MqttCaCert, string()}});

      AddDefaultedOptions(block, m_options,
                          {{configuration::MqttHost, "localhost"s},
                           {configuration::MqttPort, 1883},
                           {configuration::MqttTls, false},
                           {configuration::AutoAvailable, false},
                           {configuration::RealTime, false},
                           {configuration::RelativeTime, false}});
      loadTopics(block, m_options);

      m_handler = m_pipeline.makeHandler();
      auto clientHandler = make_unique<ClientHandler>();

      clientHandler->m_connecting = [this](shared_ptr<MqttClient> client) {
        m_handler->m_connecting(client->getIdentity());
      };

      clientHandler->m_connected = [this](shared_ptr<MqttClient> client) {
        m_handler->m_connected(client->getIdentity());
      };

      clientHandler->m_disconnected = [this](shared_ptr<MqttClient> client) {
        m_handler->m_disconnected(client->getIdentity());
      };

      clientHandler->m_receive = [this](shared_ptr<MqttClient> client, const std::string &topic,
                                        const std::string &payload) {
        m_handler->m_processMessage(topic, payload, client->getIdentity());
      };

{'fraction_non_alphanumeric': 0.11519243754220121, 'fraction_numerical': 0.004591492234976367, 'mean_word_length': 3.009745533297239, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '238655', 'n_tokens_mistral': 2069, 'n_tokens_neox': 2042, 'n_words': 470}