<commit_before>/** Copyright (C) 2016, 2017 European Spallation Source ERIC */

#include <cassert>
#include <common/Producer.h>
#include <iostream>
#include <libs/include/gccintel.h>

Producer::Producer(std::string broker, std::string topicstr) : ProducerBase() {

  conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  if (conf == nullptr) {
    std::cerr << "Unable to create CONF_GLOBAL object" << std::endl;
    return;
  }

  if (tconf == nullptr) {
    std::cerr << "Unable to create CONF_TOPIC object" << std::endl;
    return;
  }

  conf->set("metadata.broker.list", broker, errstr);
  assert(errstr.empty() == true);
  conf->set("message.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);
  conf->set("fetch.message.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);
  conf->set("message.copy.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);

  producer = RdKafka::Producer::create(conf, errstr);
  if (!producer) {
    std::cerr << "Failed to create producer: " << errstr << std::endl;
    /** @fixme add logging to Greylog */
    return;
  }

  topic = RdKafka::Topic::create(producer, topicstr, tconf, errstr);
  if (!topic) {
    std::cerr << "Failed to create topic: " << errstr << std::endl;
    /** @fixme add logging to Greylog */
    return;
  }
}

Producer::~Producer() {
  delete topic;
  delete producer;<commit_msg>First stab at improving Kafka throughput.<commit_after>/** Copyright (C) 2016, 2017 European Spallation Source ERIC */

#include <cassert>
#include <common/Producer.h>
#include <iostream>
#include <libs/include/gccintel.h>

Producer::Producer(std::string broker, std::string topicstr) : ProducerBase() {

  conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
  tconf = RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC);

  if (conf == nullptr) {
    std::cerr << "Unable to create CONF_GLOBAL object" << std::endl;
    return;
  }

  if (tconf == nullptr) {
    std::cerr << "Unable to create CONF_TOPIC object" << std::endl;
    return;
  }

  conf->set("metadata.broker.list", broker, errstr);
  assert(errstr.empty() == true);
  conf->set("message.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);
  conf->set("fetch.message.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);
  conf->set("message.copy.max.bytes", "10000000", errstr);
  assert(errstr.empty() == true);
  conf->set("queue.buffering.max.ms", "500", errstr);
  assert(errstr.empty() == true);
  conf->set("batch.num.messages", "10000", errstr);
  assert(errstr.empty() == true);

  producer = RdKafka::Producer::create(conf, errstr);
  if (!producer) {
    std::cerr << "Failed to create producer: " << errstr << std::endl;
    /** @fixme add logging to Greylog */
    return;
  }

  topic = RdKafka::Topic::create(producer, topicstr, tconf, errstr);
  if (!topic) {
    std::cerr << "Failed to create topic: " << errstr << std::endl;
    /** @fixme add logging to Greylog */
    return;
  }
}

Producer::~Producer() {
  delete topic;
  delete producer;
{'fraction_non_alphanumeric': 0.18497109826589594, 'fraction_numerical': 0.023121387283236993, 'mean_word_length': 4.389273356401384, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1530917', 'n_tokens_mistral': 1155, 'n_tokens_neox': 1082, 'n_words': 236}