<commit_before>/*
** Copyright 2017 Centreon
**
** 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.
**
** For more information : <EMAIL>
*/
#include <algorithm>
#include <cmath>
#include <sstream>
#include <sstream>
#include "com/centreon/broker/exceptions/shutdown.hh"
#include "com/centreon/broker/io/events.hh"
#include "com/centreon/broker/logging/logging.hh"
#include "com/centreon/broker/log_v2.hh"
#include "com/centreon/broker/logging/logging.hh"
#include "com/centreon/broker/lua/luabinding.hh"
#include "com/centreon/broker/lua/stream.hh"
#include "com/centreon/broker/misc/math.hh"

using namespace com::centreon::broker;
using namespace com::centreon::broker::misc;
using namespace com::centreon::broker::lua;

/**************************************
 *                                     *
 *           Public Methods            *
 *                                     *
 **************************************/

/**
 *  Constructor.
 *
 *  @param[in] addr                    Address to connect to
 *  @param[in] port                    port
 */
stream::stream(std::string const& lua_script,
               std::map<std::string, misc::variant> const& conf_params,
               std::shared_ptr<persistent_cache> const& cache)
    : _cache{cache},
      _acks_count{0},
      _stats{{}},
      _stats_it{_stats.begin()},
      _next_stat{time(nullptr)},
      _nb_stats{0},
      _a_min{1},
      _exit{false},
      _flush{false} {

  bool fail = false;
  std::string fail_msg;

  /* The lua interpreter does not support exchanges with several threads from
   * the outside. By design, the filter is called from another thread than the
   * one used for the write function.
   * To fulfill this difficulty, the filter and write functions feed this thread
   * with their arguments. The filter waits for an answer whereas the write
   * function just increases an _acks_count to inform broker on treated events.
   */
  std::unique_lock<std::mutex> lock(_loop_m);

  _thread = std::thread([&] {
    // Access to the Lua interpreter
    luabinding* lb = nullptr;
    bool has_flush = false;

    {
      std::lock_guard<std::mutex> lock(_loop_m);
      try {
        lb = new luabinding(lua_script, conf_params, _cache);
        has_flush = lb->has_flush();
      }
      catch (std::exception const& e) {
        fail_msg = e.what();
        fail = true;
        _exit = true;
        _loop_cv.notify_one();
        return;
      }
      _loop_cv.notify_one();
    }

    std::unique_lock<std::mutex> lock(_loop_m);

    for (;;) {
      log_v2::lua()->trace("stream: waiting for an event...");
       if (has_flush) {
         _loop_cv.wait(lock, [this] { return _exit || _flush || !_events.empty(); });

         if (_flush) {
           log_v2::lua()->debug("stream: flush event");
           lock.unlock();
           int32_t res = lb->flush();
           {
             std::lock_guard<std::mutex> lock(_acks_count_m);
             log_v2::lua()->trace("stream: {} events acknowledged by the script flush", res);
             _acks_count += res;
             log_v2::lua()->debug("stream: events to ack size: {}", _acks_count);
           }
           _flush = false;
           lock.lock();
         }
       }
       else
         _loop_cv.wait(lock, [this] { return _exit || !_events.empty(); });

      if (!_events.empty()) {
        log_v2::lua()->debug("stream: there are events to send to lua");
        std::shared_ptr<io::data>& d = _events.front();
        _events.pop_front();
        lock.unlock();
        uint32_t res = lb->write(d);
        {
          std::lock_guard<std::mutex> lock(_acks_count_m);
          log_v2::lua()->trace("stream: {} events acknowledged by the script write", res);
          _acks_count += res;
          log_v2::lua()->debug("stream: events to ack size: {}", _acks_count);
        }
        lock.lock();
      }<commit_msg>fix(lua): potential dangling pointer removed (#402)<commit_after>/*
** Copyright 2017 Centreon
**
** 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.
**
** For more information : <EMAIL>
*/
#include <algorithm>
#include <cmath>
#include <sstream>
#include <sstream>
#include "com/centreon/broker/exceptions/shutdown.hh"
#include "com/centreon/broker/io/events.hh"
#include "com/centreon/broker/logging/logging.hh"
#include "com/centreon/broker/log_v2.hh"
#include "com/centreon/broker/logging/logging.hh"
#include "com/centreon/broker/lua/luabinding.hh"
#include "com/centreon/broker/lua/stream.hh"
#include "com/centreon/broker/misc/math.hh"

using namespace com::centreon::broker;
using namespace com::centreon::broker::misc;
using namespace com::centreon::broker::lua;

/**************************************
 *                                     *
 *           Public Methods            *
 *                                     *
 **************************************/

/**
 *  Constructor.
 *
 *  @param[in] addr                    Address to connect to
 *  @param[in] port                    port
 */
stream::stream(std::string const& lua_script,
               std::map<std::string, misc::variant> const& conf_params,
               std::shared_ptr<persistent_cache> const& cache)
    : _cache{cache},
      _acks_count{0},
      _stats{{}},
      _stats_it{_stats.begin()},
      _next_stat{time(nullptr)},
      _nb_stats{0},
      _a_min{1},
      _exit{false},
      _flush{false} {

  bool fail = false;
  std::string fail_msg;

  /* The lua interpreter does not support exchanges with several threads from
   * the outside. By design, the filter is called from another thread than the
   * one used for the write function.
   * To fulfill this difficulty, the filter and write functions feed this thread
   * with their arguments. The filter waits for an answer whereas the write
   * function just increases an _acks_count to inform broker on treated events.
   */
  std::unique_lock<std::mutex> lock(_loop_m);

  _thread = std::thread([&] {
    // Access to the Lua interpreter
    luabinding* lb = nullptr;
    bool has_flush = false;

    {
      std::lock_guard<std::mutex> lock(_loop_m);
      try {
        lb = new luabinding(lua_script, conf_params, _cache);
        has_flush = lb->has_flush();
      }
      catch (std::exception const& e) {
        fail_msg = e.what();
        fail = true;
        _exit = true;
        _loop_cv.notify_one();
        return;
      }
      _loop_cv.notify_one();
    }

    std::unique_lock<std::mutex> lock(_loop_m);

    for (;;) {
      log_v2::lua()->trace("stream: waiting for an event...");
       if (has_flush) {
         _loop_cv.wait(lock, [this] { return _exit || _flush || !_events.empty(); });

         if (_flush) {
           log_v2::lua()->debug("stream: flush event");
           lock.unlock();
           int32_t res = lb->flush();
           {
             std::lock_guard<std::mutex> lock(_acks_count_m);
             log_v2::lua()->trace("stream: {} events acknowledged by the script flush", res);
             _acks_count += res;
             log_v2::lua()->debug("stream: events to ack size: {}", _acks_count);
           }
           _flush = false;
           lock.lock();
         }
       }
       else
         _loop_cv.wait(lock, [this] { return _exit || !_events.empty(); });

      if (!_events.empty()) {
        log_v2::lua()->debug("stream: there are events to send to lua");
        std::shared_ptr<io::data> d = _events.front();
        _events.pop_front();
        lock.unlock();
        uint32_t res = lb->write(d);
        {
          std::lock_guard<std::mutex> lock(_acks_count_m);
          log_v2::lua()->trace("stream: {} events acknowledged by the script write", res);
          _acks_count += res;
          log_v2::lua()->debug("stream: events to ack size: {}", _acks_count);
        }
        lock.lock();
      }
{'fraction_non_alphanumeric': 0.1479759609933099, 'fraction_numerical': 0.005556185508561061, 'mean_word_length': 2.6, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6768508', 'n_tokens_mistral': 2684, 'n_tokens_neox': 2595, 'n_words': 699}