src/KafkaHandler.h
author František Kučera <franta-hg@frantovo.cz>
Wed, 27 Apr 2022 23:09:36 +0200
branchv_0
changeset 1 d7824971fd9e
parent 0 f4d7e0965055
permissions -rw-r--r--
first version of Kafka producer
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     1
/**
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     2
 * Relational pipes
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     3
 * Copyright © 2022 František Kučera (Frantovo.cz, GlobalCode.info)
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     4
 *
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     5
 * This program is free software: you can redistribute it and/or modify
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     6
 * it under the terms of the GNU General Public License as published by
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     7
 * the Free Software Foundation, version 3 of the License.
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     8
 *
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
     9
 * This program is distributed in the hope that it will be useful,
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    10
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    11
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    12
 * GNU General Public License for more details.
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    13
 *
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    14
 * You should have received a copy of the GNU General Public License
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    15
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    16
 */
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    17
#pragma once
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    18
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    19
#include <memory>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    20
#include <string>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    21
#include <vector>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    22
#include <iostream>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    23
#include <sstream>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    24
#include <locale>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    25
#include <codecvt>
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    26
#include <unistd.h>
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    27
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    28
#include <librdkafka/rdkafkacpp.h>
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    29
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    30
#include <relpipe/common/type/typedefs.h>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    31
#include <relpipe/reader/TypeId.h>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    32
#include <relpipe/reader/handlers/RelationalReaderStringHandler.h>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    33
#include <relpipe/reader/handlers/AttributeMetadata.h>
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    34
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    35
#include "Configuration.h"
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    36
#include "Hex.h"
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    37
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    38
namespace relpipe {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    39
namespace out {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    40
namespace kafka {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    41
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    42
static void check(RdKafka::Conf::ConfResult result, const std::string& errString) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    43
	if (result != RdKafka::Conf::CONF_OK) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    44
		throw std::logic_error("Unable to configure Kafka: " + errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    45
	}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    46
}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    47
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    48
static void check(RdKafka::ErrorCode result, const std::string& errString) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    49
	if (result != RdKafka::ERR_NO_ERROR) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    50
		throw std::logic_error("Kafka error: " + errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    51
	}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    52
}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    53
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    54
static void check(std::shared_ptr<RdKafka::Producer> kafkaProducer, const std::string& errString) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    55
	if (kafkaProducer.get() == nullptr) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    56
		throw std::logic_error("Unable to create Kafka producer: " + errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    57
	}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    58
}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    59
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    60
class KafkaHandler : public relpipe::reader::handlers::RelationalReaderStringHandler {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    61
private:
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    62
	std::wstring_convert<codecvt_utf8<wchar_t>> convertor; // TODO: support also other encodings.
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    63
	Configuration configuration;
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    64
	std::shared_ptr<RdKafka::Producer> kafkaProducer;
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    65
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    66
	struct CurrentRelation {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    67
		relpipe::common::type::StringX name;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    68
		std::vector<relpipe::reader::handlers::AttributeMetadata> attributes;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    69
		relpipe::common::type::Integer attributeIndex = 0;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    70
		std::string currentValue;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    71
	} currentRelation;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
    72
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    73
	void connect() {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    74
		if (!kafkaProducer) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    75
			std::string errString;
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    76
			std::shared_ptr<RdKafka::Conf> producerConf(RdKafka::Conf::create(RdKafka::Conf::ConfType::CONF_GLOBAL));
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    77
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    78
			// TODO: configurable groupId, clientId and other parameters
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    79
			std::string groupId = "relpipe-in-kafka-group-" + std::to_string(getpid());
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    80
			std::string clientId = "relpipe-in-kafka-client-" + std::to_string(getpid());
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    81
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    82
			check(producerConf->set("client.id", clientId, errString), errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    83
			check(producerConf->set("group.id", groupId, errString), errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    84
			check(producerConf->set("bootstrap.servers", "plaintext://127.0.0.1:9092", errString), errString); // "host1:9092,host2:9092"    "192.168.1.56:9092"
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    85
			//check(consumerConf->set("debug", "all", errString), errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    86
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    87
			kafkaProducer.reset(RdKafka::Producer::create(producerConf.get(), errString));
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    88
			check(kafkaProducer, errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    89
		}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    90
	}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    91
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    92
	void sendMessage(const std::string& payload) {
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    93
		std::string errString;
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    94
		check(kafkaProducer->produce(
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    95
				convertor.to_bytes(configuration.queue),
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    96
				RdKafka::Topic::PARTITION_UA,
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    97
				RdKafka::Producer::RK_MSG_COPY,
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    98
				const_cast<char *> (payload.c_str()), payload.size(),
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
    99
				nullptr, 0,
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   100
				0,
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   101
				nullptr), errString);
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   102
	}
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   103
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   104
public:
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   105
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   106
	KafkaHandler(Configuration configuration) : configuration(configuration) {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   107
	}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   108
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   109
	void startRelation(relpipe::common::type::StringX name, std::vector<relpipe::reader::handlers::AttributeMetadata> attributes) override {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   110
		currentRelation = CurrentRelation{name, attributes};
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   111
		connect();
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   112
	}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   113
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   114
	void attribute(const relpipe::common::type::StringX& value) override {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   115
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   116
		auto attributeName = currentRelation.attributes[currentRelation.attributeIndex].getAttributeName();
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   117
		if (attributeName == L"text") currentRelation.currentValue = convertor.to_bytes(value);
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   118
		else if (attributeName == L"data") currentRelation.currentValue = Hex::fromHex(value).str();
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   119
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   120
		currentRelation.attributeIndex++;
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   121
		if (currentRelation.attributeIndex == currentRelation.attributes.size()) {
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   122
			currentRelation.attributeIndex = 0;
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   123
			sendMessage(currentRelation.currentValue);
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   124
		}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   125
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   126
	}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   127
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   128
	void endOfPipe() {
1
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   129
		std::string errString;
d7824971fd9e first version of Kafka producer
František Kučera <franta-hg@frantovo.cz>
parents: 0
diff changeset
   130
		check(kafkaProducer->flush(60000), errString);
0
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   131
	}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   132
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   133
};
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   134
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   135
}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   136
}
f4d7e0965055 establish project
František Kučera <franta-hg@frantovo.cz>
parents:
diff changeset
   137
}