--- a/src/DeserializeHandler.h Sun Apr 17 21:24:32 2022 +0200
+++ b/src/DeserializeHandler.h Tue Apr 19 23:17:15 2022 +0200
@@ -32,6 +32,7 @@
#include <relpipe/cli/RelpipeCLIException.h>
#include "Configuration.h"
+#include "UnionAllHandler.h"
namespace relpipe {
namespace tr {
@@ -41,18 +42,84 @@
private:
Configuration configuration;
shared_ptr<relpipe::writer::RelationalWriter> writer;
+ std::wstring_convert<codecvt_utf8<wchar_t>> convertor;
+ UnionAllHandler unionAllHandler;
+ class RelationContext {
+ public:
+ relpipe::common::type::StringX name;
+ std::vector<relpipe::reader::handlers::AttributeMetadata> readerMetadata;
+ std::vector<relpipe::writer::AttributeMetadata> writerMetadata;
+ } relationContext;
+
+ class RecordContext {
+ public:
+ std::stringstream buffer;
+ size_t attributeIndex = 0;
+ } recordContext;
+
+ char fromHex(wchar_t ch) {
+ if (L'0' <= ch && ch <= L'9') return ch - L'0';
+ else if (L'a' <= ch && ch <= L'f') return ch - L'a' + 10;
+ else throw relpipe::reader::RelpipeReaderException(L"Unable to decode hexadeximal string.");
+ }
+
+ std::stringstream fromHex(const relpipe::common::type::StringX& hex) {
+ std::stringstream octets;
+
+ char octet = 0;
+
+ for (size_t i = 0, limit = hex.size(); i < limit; i++) {
+ if (i % 2 == 0) {
+ octet = fromHex(hex[i]) << 4;
+ } else {
+ octet += fromHex(hex[i]);
+ octets.put(octet);
+ }
+ }
+
+ return octets;
+ }
+
public:
- DeserializeHandler(shared_ptr<relpipe::writer::RelationalWriter> writer, Configuration configuration) : writer(writer), configuration(configuration) {
+ DeserializeHandler(shared_ptr<relpipe::writer::RelationalWriter> writer, Configuration configuration) : writer(writer), configuration(configuration), unionAllHandler(writer, configuration) {
+ // TODO: configurable relation name?
+ // TODO: configurable attribute name?
+ // TODO: optional custom attributes with constant value or ordinal number?
+ // TODO: optional deserialization of only certain relations? and certain fields?
+ // TODO: optional pass-through of certain relations?
}
virtual ~DeserializeHandler() = default;
void startRelation(relpipe::common::type::StringX name, std::vector<relpipe::reader::handlers::AttributeMetadata> attributes) override {
+ relationContext = RelationContext();
+
+ relationContext.name = name;
+ relationContext.readerMetadata = attributes;
+
+ for (relpipe::reader::handlers::AttributeMetadata readerMetadata : attributes) {
+ relationContext.writerMetadata.push_back({readerMetadata.getAttributeName(), writer->toTypeId(readerMetadata.getTypeName())});
+ }
}
void attribute(const relpipe::common::type::StringX& value) override {
+ auto attributeName = relationContext.readerMetadata[recordContext.attributeIndex].getAttributeName();
+
+ if (recordContext.attributeIndex == 0) recordContext = RecordContext();
+
+ if (attributeName == L"data") recordContext.buffer = fromHex(value);
+
+ recordContext.attributeIndex++;
+
+ if (recordContext.attributeIndex % relationContext.readerMetadata.size() == 0) {
+ std::stringstream octets = fromHex(value); // TODO: skip this step once we have octet-string data type + typed values instead of strings
+ std::shared_ptr<relpipe::reader::RelationalReader> reader(relpipe::reader::Factory::create(octets));
+ reader->addHandler(&unionAllHandler);
+ reader->process();
+ recordContext.attributeIndex = 0;
+ }
}
void endOfPipe() {