tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Initial version of new bolt encoder.

Browse Source 
Reviewers: dgleich, buda

Reviewed By: buda

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D104
This commit is contained in:
Matej Ferencevic 2017-03-22 15:53:30 +01:00
parent f505c76189
commit 3bf0bd40a7
11 changed files with 912 additions and 210 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
CMakeLists.txt
View File

@ -255,6 +255,8 @@ option(MANUAL_TESTS "Add manual test binaries" OFF)
message(STATUS "Add manual test binaries: ${MANUAL_TESTS}")
option(UNIT_TESTS "Add unit test binaries" OFF)
message(STATUS "Add unit test binaries: ${UNIT_TESTS}")
option(HARDCODED_TARGETS "Make hardcoded query targets" ON)
message(STATUS "Make hardcoded query targets: ${HARDCODED_TARGETS}")
option(TEST_COVERAGE "Generate coverage reports from unit tests" OFF)
message(STATUS "Generate coverage from unit tests: ${TEST_COVERAGE}")
# -----------------------------------------------------------------------------
@ -451,21 +453,23 @@ set_target_properties(__refactor_target PROPERTIES EXCLUDE_FROM_ALL 1)
# 2. query plan execution agains empty database and injected OutputStream
# 3. integration tests for all pilot/clients written in cucumber
# the following targets address only the first phase
file(GLOB __HARDCODED_SOURCES
${CMAKE_SOURCE_DIR}/tests/integration/hardcoded_query/*.cpp)
foreach(file_path ${__HARDCODED_SOURCES})
get_filename_component(file_name ${file_path} NAME_WE)
set(target_name __${file_name}_hardcoded_target)
add_executable(${target_name} ${CMAKE_SOURCE_DIR}/libs/__main.cpp
${file_path})
target_link_libraries(${target_name} memgraph_lib)
target_link_libraries(${target_name} fmt)
target_link_libraries(${target_name} Threads::Threads)
set_property(TARGET ${target_name} PROPERTY CXX_STANDARD ${cxx_standard})
set_target_properties(${target_name}
PROPERTIES RUNTIME_OUTPUT_DIRECTORY
"${CMAKE_BINARY_DIR}/__hardcoded_targets")
endforeach()
if(HARDCODED_TARGETS)
file(GLOB __HARDCODED_SOURCES
${CMAKE_SOURCE_DIR}/tests/integration/hardcoded_query/*.cpp)
foreach(file_path ${__HARDCODED_SOURCES})
get_filename_component(file_name ${file_path} NAME_WE)
set(target_name __${file_name}_hardcoded_target)
add_executable(${target_name} ${CMAKE_SOURCE_DIR}/libs/__main.cpp
${file_path})
target_link_libraries(${target_name} memgraph_lib)
target_link_libraries(${target_name} fmt)
target_link_libraries(${target_name} Threads::Threads)
set_property(TARGET ${target_name} PROPERTY CXX_STANDARD ${cxx_standard})
set_target_properties(${target_name}
PROPERTIES RUNTIME_OUTPUT_DIRECTORY
"${CMAKE_BINARY_DIR}/__hardcoded_targets")
endforeach()
endif()
get_target_cxx_flags(memgraph_lib compile_flags)
set(plan_compiler_flags_file ${build_include_dir}/query/plan_compiler_flags.hpp)

2
coverage
View File

@ -19,7 +19,7 @@ all_coverage_info=""
for dir in $binary_path/*.dir; do
pushd ${dir}
lcov --gcov-tool ${working_dir}/llvm-gcov -c -d . -o ${coverage_file}
lcov -r ${coverage_file} '/usr/*' '*/libs/*' -o ${coverage_file}
lcov -r ${coverage_file} '/usr/*' '*/libs/*' '*/tests/*' -o ${coverage_file}
all_coverage_info+=" -a ${working_dir}/${dir}/${coverage_file}"
popd
done

80
src/communication/bolt/v1/encoder/chunked_buffer.hpp Normal file
View File

@ -0,0 +1,80 @@
#pragma once
#include <cstring>
#include <memory>
#include <vector>
#include <algorithm>
#include "communication/bolt/v1/config.hpp"
#include "logging/default.hpp"
#include "utils/types/byte.hpp"
#include "utils/bswap.hpp"
namespace communication::bolt {
// maximum chunk size = 65536 bytes data
static constexpr size_t CHUNK_SIZE = 65536;
/**
* Bolt chunked buffer.
* Has methods for writing and flushing data.
* Writing data stores data in the internal buffer and flushing data sends
* the currently stored data to the Socket with prepended data length and
* appended chunk tail (0x00 0x00).
*
* @tparam Socket the output socket that should be used
*/
template <class Socket>
class ChunkedBuffer {
public:
ChunkedBuffer(Socket &socket) : socket_(socket), logger_(logging::log->logger("Chunked Buffer")) {}
void Write(const uint8_t* values, size_t n) {
logger_.trace("Write {} bytes", n);
// total size of the buffer is now bigger for n
size_ += n;
// reserve enough space for the new data
buffer_.reserve(size_);
// copy new data
std::copy(values, values + n, std::back_inserter(buffer_));
}
void Flush() {
size_t size = buffer_.size(), n = 0, pos = 0;
uint16_t head;
while (size > 0) {
head = n = std::min(CHUNK_SIZE, size);
head = bswap(head);
logger_.trace("Flushing chunk of {} bytes", n);
// TODO: implement better flushing strategy!
socket_.Write(reinterpret_cast<const uint8_t *>(&head), sizeof(head));
socket_.Write(buffer_.data() + pos, n);
head = 0;
socket_.Write(reinterpret_cast<const uint8_t *>(&head), sizeof(head));
size -= n;
pos += n;
}
// GC
// TODO: impelement a better strategy
buffer_.clear();
// clear size
size_ = 0;
}
private:
Socket& socket_;
Logger logger_;
std::vector<uint8_t> buffer_;
size_t size_{0};
};
}

312
src/communication/bolt/v1/encoder/encoder.hpp Normal file
View File

@ -0,0 +1,312 @@
#pragma once
#include "database/graph_db_accessor.hpp"
#include "logging/default.hpp"
#include "query/backend/cpp/typed_value.hpp"
#include "utils/bswap.hpp"
#include <string>
namespace communication::bolt {
static constexpr uint8_t TSTRING = 0, TLIST = 1, TMAP = 2;
static constexpr uint8_t type_tiny_marker[3] = { 0x80, 0x90, 0xA0 };
static constexpr uint8_t type_8_marker[3] = { 0xD0, 0xD4, 0xD8 };
static constexpr uint8_t type_16_marker[3] = { 0xD1, 0xD5, 0xD9 };
static constexpr uint8_t type_32_marker[3] = { 0xD2, 0xD6, 0xDA };
/**
* Bolt Encoder.
* Has public interfaces for writing Bolt specific response messages.
* Supported messages are: Record, Success, Failure and Ignored.
*
* @tparam Buffer the output buffer that should be used
* @tparam Socket the output socket that should be used
*/
template <typename Buffer, typename Socket>
class Encoder {
public:
Encoder(Socket& socket) : socket_(socket), buffer_(socket), logger_(logging::log->logger("communication::bolt::Encoder")) {}
/**
* Sends a Record message.
*
* From the Bolt v1 documentation:
* RecordMessage (signature=0x71) {
* List<Value> fields
* }
*
* @param values the fields list object that should be sent
*/
void MessageRecord(const std::vector<TypedValue>& values) {
// 0xB1 = struct 1; 0x71 = record signature
WriteRAW("\xB1\x71", 2);
WriteList(values);
buffer_.Flush();
}
/**
* Sends a Success message.
*
* From the Bolt v1 documentation:
* SuccessMessage (signature=0x70) {
* Map<String,Value> metadata
* }
*
* @param metadata the metadata map object that should be sent
*/
void MessageSuccess(const std::map<std::string, TypedValue>& metadata) {
// 0xB1 = struct 1; 0x70 = success signature
WriteRAW("\xB1\x70", 2);
WriteMap(metadata);
buffer_.Flush();
}
/**
* Sends a Failure message.
*
* From the Bolt v1 documentation:
* FailureMessage (signature=0x7F) {
* Map<String,Value> metadata
* }
*
* @param metadata the metadata map object that should be sent
*/
void MessageFailure(const std::map<std::string, TypedValue>& metadata) {
// 0xB1 = struct 1; 0x7F = failure signature
WriteRAW("\xB1\x7F", 2);
WriteMap(metadata);
buffer_.Flush();
}
/**
* Sends an Ignored message.
*
* From the bolt v1 documentation:
* IgnoredMessage (signature=0x7E) {
* Map<String,Value> metadata
* }
*
* @param metadata the metadata map object that should be sent
*/
void MessageIgnored(const std::map<std::string, TypedValue>& metadata) {
// 0xB1 = struct 1; 0x7E = ignored signature
WriteRAW("\xB1\x7E", 2);
WriteMap(metadata);
buffer_.Flush();
}
/**
* Sends an Ignored message.
*
* This function sends an ignored message without additional metadata.
*/
void MessageIgnored() {
// 0xB0 = struct 0; 0x7E = ignored signature
WriteRAW("\xB0\x7E", 2);
buffer_.Flush();
}
private:
Socket& socket_;
Buffer buffer_;
Logger logger_;
void WriteRAW(const uint8_t* data, uint64_t len) {
buffer_.Write(data, len);
}
void WriteRAW(const char* data, uint64_t len) {
WriteRAW((const uint8_t*) data, len);
}
void WriteRAW(const uint8_t data) {
WriteRAW(&data, 1);
}
template <class T>
void WriteValue(T value) {
value = bswap(value);
WriteRAW(reinterpret_cast<const uint8_t *>(&value), sizeof(value));
}
void WriteNull() {
// 0xC0 = null marker
WriteRAW(0xC0);
}
void WriteBool(const bool& value) {
if (value) {
// 0xC3 = true marker
WriteRAW(0xC3);
} else {
// 0xC2 = false marker
WriteRAW(0xC2);
}
}
void WriteInt(const int64_t& value) {
if (value >= -16L && value < 128L) {
WriteRAW(static_cast<uint8_t>(value));
} else if (value >= -128L && value < -16L) {
// 0xC8 = int8 marker
WriteRAW(0xC8);
WriteRAW(static_cast<uint8_t>(value));
} else if (value >= -32768L && value < 32768L) {
// 0xC9 = int16 marker
WriteRAW(0xC9);
WriteValue(static_cast<int16_t>(value));
} else if (value >= -2147483648L && value < 2147483648L) {
// 0xCA = int32 marker
WriteRAW(0xCA);
WriteValue(static_cast<int32_t>(value));
} else {
// 0xCB = int64 marker
WriteRAW(0xCB);
WriteValue(value);
}
}
void WriteDouble(const double& value) {
// 0xC1 = float64 marker
WriteRAW(0xC1);
WriteValue(*reinterpret_cast<const int64_t *>(&value));
}
void WriteTypeSize(const size_t size, const uint8_t typ) {
if (size <= 15) {
uint8_t len = size;
len &= 0x0F;
// tiny marker (+len)
WriteRAW(type_tiny_marker[typ] + len);
} else if (size <= 255) {
uint8_t len = size;
// 8 marker
WriteRAW(type_8_marker[typ]);
WriteRAW(len);
} else if (size <= 65536) {
uint16_t len = size;
// 16 marker
WriteRAW(type_16_marker[typ]);
WriteValue(len);
} else {
uint32_t len = size;
// 32 marker
WriteRAW(type_32_marker[typ]);
WriteValue(len);
}
}
void WriteString(const std::string& value) {
WriteTypeSize(value.size(), TSTRING);
WriteRAW(value.c_str(), value.size());
}
void WriteList(const std::vector<TypedValue>& value) {
WriteTypeSize(value.size(), TLIST);
for (auto& x: value) WriteTypedValue(x);
}
void WriteMap(const std::map<std::string, TypedValue>& value) {
WriteTypeSize(value.size(), TMAP);
for (auto& x: value) {
WriteString(x.first);
WriteTypedValue(x.second);
}
}
void WriteVertex(const VertexAccessor& vertex) {
// 0xB3 = struct 3; 0x4E = vertex signature
WriteRAW("\xB3\x4E", 2);
// IMPORTANT: here we write a hardcoded 0 because we don't
// use internal IDs, but need to give something to Bolt
// note that OpenCypher has no id(x) function, so the client
// should not be able to do anything with this value anyway
WriteInt(0);
// write labels
const auto& labels = vertex.labels();
WriteTypeSize(labels.size(), TLIST);
for (const auto& label : labels)
WriteString(vertex.db_accessor().label_name(label));
// write properties
const auto& props = vertex.Properties();
WriteTypeSize(props.size(), TMAP);
for (const auto& prop : props) {
WriteString(vertex.db_accessor().property_name(prop.first));
WriteTypedValue(prop.second);
}
}
void WriteEdge(const EdgeAccessor& edge) {
// 0xB5 = struct 5; 0x52 = edge signature
WriteRAW("\xB5\x52", 2);
// IMPORTANT: here we write a hardcoded 0 because we don't
// use internal IDs, but need to give something to Bolt
// note that OpenCypher has no id(x) function, so the client
// should not be able to do anything with this value anyway
WriteInt(0);
WriteInt(0);
WriteInt(0);
// write type
WriteString(edge.db_accessor().edge_type_name(edge.edge_type()));
// write properties
const auto& props = edge.Properties();
WriteTypeSize(props.size(), TMAP);
for (const auto& prop : props) {
WriteString(edge.db_accessor().property_name(prop.first));
WriteTypedValue(prop.second);
}
}
void WritePath() {
// TODO: this isn't implemented in the backend!
}
void WriteTypedValue(const TypedValue& value) {
switch(value.type()) {
case TypedValue::Type::Null:
WriteNull();
break;
case TypedValue::Type::Bool:
WriteBool(value.Value<bool>());
break;
case TypedValue::Type::Int:
WriteInt(value.Value<int64_t>());
break;
case TypedValue::Type::Double:
WriteDouble(value.Value<double>());
break;
case TypedValue::Type::String:
WriteString(value.Value<std::string>());
break;
case TypedValue::Type::List:
WriteList(value.Value<std::vector<TypedValue>>());
break;
case TypedValue::Type::Map:
WriteMap(value.Value<std::map<std::string, TypedValue>>());
break;
case TypedValue::Type::Vertex:
WriteVertex(value.Value<VertexAccessor>());
break;
case TypedValue::Type::Edge:
WriteEdge(value.Value<EdgeAccessor>());
break;
case TypedValue::Type::Path:
// TODO: this is not implemeted yet!
WritePath();
break;
}
}
};
}

82
src/communication/bolt/v1/encoder/result_stream.hpp Normal file
View File

@ -0,0 +1,82 @@
#pragma once
#include "communication/bolt/v1/encoder/encoder.hpp"
#include "communication/bolt/v1/encoder/chunked_buffer.hpp"
#include "query/backend/cpp/typed_value.hpp"
#include "logging/default.hpp"
namespace communication::bolt {
/**
* A high level API for streaming a Bolt response. Exposes
* functionalities used by the compiler and query plans (which
* should not use any lower level API).
*
* @tparam Socket Socket used.
*/
template <typename Socket>
class ResultStream {
private:
using encoder_t = Encoder<ChunkedBuffer<Socket>, Socket>;
public:
// TODO add logging to this class
ResultStream(encoder_t &encoder) :
encoder_(encoder) {}
/**
* Writes a header. Typically a header is something like:
* [
* "Header1",
* "Header2",
* "Header3"
* ]
*
* @param fields the header fields that should be sent.
*/
void Header(const std::vector<std::string> &fields) {
std::vector<TypedValue> vec;
std::map<std::string, TypedValue> data;
for (auto& i : fields)
vec.push_back(TypedValue(i));
data.insert(std::make_pair(std::string("fields"), TypedValue(vec)));
encoder_.MessageSuccess(data);
}
/**
* Writes a result. Typically a result is something like:
* [
* Value1,
* Value2,
* Value3
* ]
* NOTE: The result fields should be in the same ordering that the header
* fields were sent in.
*
* @param values the values that should be sent
*/
void Result(std::vector<TypedValue> &values) {
encoder_.MessageRecord(values);
}
/**
* Writes a summary. Typically a summary is something like:
* {
* "type" : "r" | "rw" | ...,
* "stats": {
* "nodes_created": 12,
* "nodes_deleted": 0
* }
* }
*
* @param summary the summary map object that should be sent
*/
void Summary(const std::map<std::string, TypedValue> &summary) {
encoder_.MessageSuccess(summary);
}
private:
encoder_t& encoder_;
};
}

137
src/communication/bolt/v1/serialization/result_stream.hpp
View File

@ -1,137 +0,0 @@
#pragma once
#include "communication/bolt/v1/serialization/bolt_serializer.hpp"
#include "query/backend/cpp/typed_value.hpp"
#include "logging/default.hpp"
namespace bolt {
/**
* A high level API for streaming a Bolt response. Exposes
* functionalities used by the compiler and query plans (which
* should not use any lower level API).
*
* @tparam TChunkedEncoder Type of chunked encoder used.
*/
// TODO templatisation on TChunkedEncoder might not be desired
// but makes the code a bit easer to understand because we know
// that this class uses a BoltEncoder (and not some arbitrary template)
// it helps the programmer, the compiler and the IDE
template <typename TChunkedEncoder>
class RecordStream {
public:
// TODO add logging to this class
RecordStream(BoltEncoder<TChunkedEncoder> &bolt_encoder)
: bolt_encoder_(bolt_encoder), serializer_(bolt_encoder) {}
void Header(const std::vector<std::string> &fields) {
bolt_encoder_.message_success();
bolt_encoder_.write_map_header(1);
bolt_encoder_.write_string("fields");
bolt_encoder_.write_list_header(fields.size());
for (auto &name : fields) {
bolt_encoder_.write_string(name);
}
Chunk();
Send();
}
void Result(std::vector<TypedValue> &values) {
bolt_encoder_.message_record();
Write(values);
Chunk();
Send();
}
/**
* Writes a summary. Typically a summary is something like:
* {
* "type" : "r" | "rw" | ...,
* "stats": {
* "nodes_created": 12,
* "nodes_deleted": 0
* }
* }
*
* @param value
*/
void Summary(const std::map<std::string, TypedValue> &summary) {
bolt_encoder_.message_success();
Write(summary);
Chunk();
}
private:
BoltEncoder<TChunkedEncoder> bolt_encoder_;
BoltSerializer<BoltEncoder<TChunkedEncoder>> serializer_;
/**
* Writes a TypedValue. Resolves it's type and uses
* encoder primitives to write exactly typed values.
*/
void Write(const TypedValue &value) {
switch (value.type()) {
case TypedValue::Type::Null:
bolt_encoder_.write_null();
break;
case TypedValue::Type::Bool:
bolt_encoder_.write(value.Value<bool>());
break;
case TypedValue::Type::Int:
bolt_encoder_.write(value.Value<int64_t>());
break;
case TypedValue::Type::Double:
bolt_encoder_.write(value.Value<double>());
break;
case TypedValue::Type::String:
bolt_encoder_.write(value.Value<std::string>());
break;
case TypedValue::Type::List:
Write(value.Value<std::vector<TypedValue>>());
break;
case TypedValue::Type::Map:
Write(value.Value<std::map<std::string, TypedValue>>());
break;
case TypedValue::Type::Vertex:
serializer_.write(value.Value<VertexAccessor>());
break;
case TypedValue::Type::Edge:
serializer_.write(value.Value<EdgeAccessor>());
break;
default:
throw std::runtime_error(
"Serialization not implemented for given type");
}
}
void Write(const std::vector<TypedValue> &values) {
bolt_encoder_.write_list_header(values.size());
for (const auto &value : values) Write(value);
}
void Write(const std::map<std::string, TypedValue> &values) {
bolt_encoder_.write_map_header(values.size());
for (const auto &kv : values) {
bolt_encoder_.write(kv.first);
Write(kv.second);
}
}
void Send() {
// TODO expose these low level functions in the encoder
// be careful! ChunkedEncoder seems to have a 'flush()' function
// but that is different from it's underlying ChunkedBuffer's
// 'flush()' method
bolt_encoder_.flush();
}
void Chunk() {
// TODO expose these low level functions in the encoder
bolt_encoder_.write_chunk();
}
};
}

46
tests/unit/bolt_chunked_buffer.cpp Normal file
View File

@ -0,0 +1,46 @@
#define NDEBUG
#include "bolt_common.hpp"
#include "communication/bolt/v1/encoder/chunked_buffer.hpp"
constexpr const int SIZE = 131072;
uint8_t data[SIZE];
void verify_output(std::vector<uint8_t>& output, const uint8_t* data, uint64_t size) {
uint64_t len = 0, pos = 0;
uint8_t tail[2] = { 0, 0 };
uint16_t head;
while (size > 0) {
head = len = std::min(size, communication::bolt::CHUNK_SIZE);
head = bswap(head);
check_output(output, reinterpret_cast<uint8_t *>(&head), sizeof(head), false);
check_output(output, data + pos, len, false);
check_output(output, tail, 2, false);
size -= len;
pos += len;
}
check_output(output, nullptr, 0, true);
}
TEST(Bolt, ChunkedBuffer) {
TestSocket socket(10);
communication::bolt::ChunkedBuffer<TestSocket> chunked_buffer(socket);
std::vector<uint8_t>& output = socket.output;
for (int i = 0; i <= SIZE; i += 16) {
chunked_buffer.Write(data, i);
chunked_buffer.Flush();
verify_output(output, data, i);
}
}
int main(int argc, char** argv) {
initialize_data(data, SIZE);
logging::init_sync();
logging::log->pipe(std::make_unique<Stdout>());
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

72
tests/unit/bolt_common.hpp Normal file
View File

@ -0,0 +1,72 @@
#include <array>
#include <cassert>
#include <cstring>
#include <iostream>
#include <vector>
#include "gtest/gtest.h"
#include "logging/default.hpp"
#include "logging/streams/stdout.hpp"
#include "dbms/dbms.hpp"
class TestSocket {
public:
TestSocket(int socket) : socket(socket) {}
TestSocket(const TestSocket& s) : socket(s.id()){};
TestSocket(TestSocket&& other) { *this = std::forward<TestSocket>(other); }
TestSocket& operator=(TestSocket&& other) {
this->socket = other.socket;
other.socket = -1;
return *this;
}
void Close() { socket = -1; }
bool IsOpen() { return socket != -1; }
int id() const { return socket; }
int Write(const std::string& str) { return Write(str.c_str(), str.size()); }
int Write(const char* data, size_t len) {
return Write(reinterpret_cast<const uint8_t*>(data), len);
}
int Write(const uint8_t* data, size_t len) {
for (int i = 0; i < len; ++i) output.push_back(data[i]);
return len;
}
std::vector<uint8_t> output;
protected:
int socket;
};
void print_output(std::vector<uint8_t>& output) {
fprintf(stderr, "output: ");
for (int i = 0; i < output.size(); ++i) {
fprintf(stderr, "%02X ", output[i]);
}
fprintf(stderr, "\n");
}
void check_output(std::vector<uint8_t>& output, const uint8_t* data,
uint64_t len, bool clear = true) {
if (clear) ASSERT_EQ(len, output.size());
else ASSERT_LE(len, output.size());
for (int i = 0; i < len; ++i)
EXPECT_EQ(output[i], data[i]);
if (clear) output.clear();
else output.erase(output.begin(), output.begin() + len);
}
void initialize_data(uint8_t* data, size_t size) {
std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<> dis(0, 255);
for (int i = 0; i < size; ++i) {
data[i] = dis(gen);
}
}

249
tests/unit/bolt_encoder.cpp Normal file
View File

@ -0,0 +1,249 @@
#include "bolt_common.hpp"
#include "communication/bolt/v1/encoder/encoder.hpp"
#include "database/graph_db.hpp"
#include "database/graph_db_accessor.hpp"
#include "query/backend/cpp/typed_value.hpp"
class TestBuffer {
public:
TestBuffer(TestSocket& socket) : socket_(socket) {}
void Write(const uint8_t* data, size_t n) {
socket_.Write(data, n);
}
void Flush() {}
private:
TestSocket& socket_;
};
const int64_t int_input[] = { 0, -1, -8, -16, 1, 63, 127, -128, -20, -17, -32768, -12345, -129, 128, 12345, 32767, -2147483648L, -12345678L, -32769L, 32768L, 12345678L, 2147483647L, -9223372036854775807L, -12345678912345L, -2147483649L, 2147483648L, 12345678912345L, 9223372036854775807 };
const uint8_t int_output[][10] = { "\x00", "\xFF", "\xF8", "\xF0", "\x01", "\x3F", "\x7F", "\xC8\x80", "\xC8\xEC", "\xC8\xEF", "\xC9\x80\x00", "\xC9\xCF\xC7", "\xC9\xFF\x7F", "\xC9\x00\x80", "\xC9\x30\x39", "\xC9\x7F\xFF", "\xCA\x80\x00\x00\x00", "\xCA\xFF\x43\x9E\xB2", "\xCA\xFF\xFF\x7F\xFF", "\xCA\x00\x00\x80\x00", "\xCA\x00\xBC\x61\x4E", "\xCA\x7F\xFF\xFF\xFF", "\xCB\x80\x00\x00\x00\x00\x00\x00\x01", "\xCB\xFF\xFF\xF4\xC5\x8C\x31\xA4\xA7", "\xCB\xFF\xFF\xFF\xFF\x7F\xFF\xFF\xFF", "\xCB\x00\x00\x00\x00\x80\x00\x00\x00", "\xCB\x00\x00\x0B\x3A\x73\xCE\x5B\x59", "\xCB\x7F\xFF\xFF\xFF\xFF\xFF\xFF\xFF" };
const uint32_t int_output_len[] = { 1, 1, 1, 1, 1, 1, 1, 2, 2, 2, 3, 3, 3, 3, 3, 3, 5, 5, 5, 5, 5, 5, 9, 9, 9, 9, 9, 9 };
const double double_input[] = { 5.834, 108.199, 43677.9882, 254524.5851 };
const uint8_t double_output[][10] = { "\xC1\x40\x17\x56\x04\x18\x93\x74\xBC", "\xC1\x40\x5B\x0C\xBC\x6A\x7E\xF9\xDB", "\xC1\x40\xE5\x53\xBF\x9F\x55\x9B\x3D", "\xC1\x41\x0F\x11\xE4\xAE\x48\xE8\xA7"};
const uint8_t vertexedge_output[] = "\xB1\x71\x93\xB3\x4E\x00\x92\x86\x6C\x61\x62\x65\x6C\x31\x86\x6C\x61\x62\x65\x6C\x32\xA2\x85\x70\x72\x6F\x70\x31\x0C\x85\x70\x72\x6F\x70\x32\xC9\x00\xC8\xB3\x4E\x00\x90\xA0\xB5\x52\x00\x00\x00\x88\x65\x64\x67\x65\x74\x79\x70\x65\xA2\x85\x70\x72\x6F\x70\x33\x2A\x85\x70\x72\x6F\x70\x34\xC9\x04\xD2";
constexpr const int SIZE = 131072;
uint8_t data[SIZE];
const uint64_t sizes[] = { 0, 1, 5, 15, 16, 120, 255, 256, 12345, 65535, 65536, 100000 };
const uint64_t sizes_num = 12;
constexpr const int STRING = 0, LIST = 1, MAP = 2;
const uint8_t type_tiny_magic[] = { 0x80, 0x90, 0xA0 };
const uint8_t type_8_magic[] = { 0xD0, 0xD4, 0xD8 };
const uint8_t type_16_magic[] = { 0xD1, 0xD5, 0xD9 };
const uint8_t type_32_magic[] = { 0xD2, 0xD6, 0xDA };
void check_type_size(std::vector<uint8_t>& v, int typ, uint64_t size) {
if (size <= 15) {
uint8_t len = size;
len &= 0x0F;
len += type_tiny_magic[typ];
check_output(v, &len, 1, false);
} else if (size <= 255) {
uint8_t len = size;
check_output(v, &type_8_magic[typ], 1, false);
check_output(v, &len, 1, false);
} else if (size <= 65536) {
uint16_t len = size;
len = bswap(len);
check_output(v, &type_16_magic[typ], 1, false);
check_output(v, reinterpret_cast<const uint8_t*> (&len), 2, false);
} else {
uint32_t len = size;
len = bswap(len);
check_output(v, &type_32_magic[typ], 1, false);
check_output(v, reinterpret_cast<const uint8_t*> (&len), 4, false);
}
}
void check_record_header(std::vector<uint8_t>& v, uint64_t size) {
check_output(v, (const uint8_t*) "\xB1\x71", 2, false);
check_type_size(v, LIST, size);
}
TestSocket socket(10);
communication::bolt::Encoder<TestBuffer, TestSocket> bolt_encoder(socket);
std::vector<uint8_t>& output = socket.output;
TEST(BoltEncoder, NullAndBool) {
std::vector<TypedValue> vals;
vals.push_back(TypedValue::Null);
vals.push_back(TypedValue(true));
vals.push_back(TypedValue(false));
bolt_encoder.MessageRecord(vals);
check_record_header(output, 3);
check_output(output, (const uint8_t*) "\xC0\xC3\xC2", 3);
}
TEST(BoltEncoder, Int) {
int N = 28;
std::vector<TypedValue> vals;
for (int i = 0; i < N; ++i)
vals.push_back(TypedValue(int_input[i]));
bolt_encoder.MessageRecord(vals);
check_record_header(output, N);
for (int i = 0; i < N; ++i)
check_output(output, int_output[i], int_output_len[i], false);
check_output(output, nullptr, 0);
}
TEST(BoltEncoder, Double) {
int N = 4;
std::vector<TypedValue> vals;
for (int i = 0; i < N; ++i)
vals.push_back(TypedValue(double_input[i]));
bolt_encoder.MessageRecord(vals);
check_record_header(output, N);
for (int i = 0; i < N; ++i)
check_output(output, double_output[i], 9, false);
check_output(output, nullptr, 0);
}
TEST(BoltEncoder, String) {
std::vector<TypedValue> vals;
for (int i = 0; i < sizes_num; ++i)
vals.push_back(TypedValue(std::string((const char*) data, sizes[i])));
bolt_encoder.MessageRecord(vals);
check_record_header(output, vals.size());
for (int i = 0; i < sizes_num; ++i) {
check_type_size(output, STRING, sizes[i]);
check_output(output, data, sizes[i], false);
}
check_output(output, nullptr, 0);
}
TEST(BoltEncoder, List) {
std::vector<TypedValue> vals;
for (int i = 0; i < sizes_num; ++i) {
std::vector<TypedValue> val;
for (int j = 0; j < sizes[i]; ++j)
val.push_back(TypedValue(std::string((const char*) &data[j], 1)));
vals.push_back(TypedValue(val));
}
bolt_encoder.MessageRecord(vals);
check_record_header(output, vals.size());
for (int i = 0; i < sizes_num; ++i) {
check_type_size(output, LIST, sizes[i]);
for (int j = 0; j < sizes[i]; ++j) {
check_type_size(output, STRING, 1);
check_output(output, &data[j], 1, false);
}
}
check_output(output, nullptr, 0);
}
TEST(BoltEncoder, Map) {
std::vector<TypedValue> vals;
uint8_t buff[10];
for (int i = 0; i < sizes_num; ++i) {
std::map<std::string, TypedValue> val;
for (int j = 0; j < sizes[i]; ++j) {
sprintf((char*) buff, "%05X", j);
std::string tmp((char*) buff, 5);
val.insert(std::make_pair(tmp, TypedValue(tmp)));
}
vals.push_back(TypedValue(val));
}
bolt_encoder.MessageRecord(vals);
check_record_header(output, vals.size());
for (int i = 0; i < sizes_num; ++i) {
check_type_size(output, MAP, sizes[i]);
for (int j = 0; j < sizes[i]; ++j) {
sprintf((char*) buff, "%05X", j);
check_type_size(output, STRING, 5);
check_output(output, buff, 5, false);
check_type_size(output, STRING, 5);
check_output(output, buff, 5, false);
}
}
check_output(output, nullptr, 0);
}
TEST(BoltEncoder, VertexAndEdge) {
// create vertex
Dbms dbms;
auto db_accessor = dbms.active();
auto va1 = db_accessor.insert_vertex();
auto va2 = db_accessor.insert_vertex();
std::string l1("label1"), l2("label2");
va1.add_label(&l1);
va1.add_label(&l2);
std::string p1("prop1"), p2("prop2");
PropertyValue pv1(12), pv2(200);
va1.PropsSet(&p1, pv1);
va1.PropsSet(&p2, pv2);
// create edge
std::string et("edgetype");
auto ea = db_accessor.insert_edge(va1, va2, &et);
std::string p3("prop3"), p4("prop4");
PropertyValue pv3(42), pv4(1234);
ea.PropsSet(&p3, pv3);
ea.PropsSet(&p4, pv4);
// check everything
std::vector<TypedValue> vals;
vals.push_back(TypedValue(va1));
vals.push_back(TypedValue(va2));
vals.push_back(TypedValue(ea));
bolt_encoder.MessageRecord(vals);
check_output(output, vertexedge_output, 74);
}
TEST(BoltEncoder, BoltV1ExampleMessages) {
// this test checks example messages from: http://boltprotocol.org/v1/
// record message
std::vector<TypedValue> rvals;
for (int i = 1; i < 4; ++i) rvals.push_back(TypedValue(i));
bolt_encoder.MessageRecord(rvals);
check_output(output, (const uint8_t*) "\xB1\x71\x93\x01\x02\x03", 6);
// success message
std::string sv1("name"), sv2("age"), sk("fields");
std::vector<TypedValue> svec;
svec.push_back(TypedValue(sv1));
svec.push_back(TypedValue(sv2));
TypedValue slist(svec);
std::map<std::string, TypedValue> svals;
svals.insert(std::make_pair(sk, slist));
bolt_encoder.MessageSuccess(svals);
check_output(output, (const uint8_t*) "\xB1\x70\xA1\x86\x66\x69\x65\x6C\x64\x73\x92\x84\x6E\x61\x6D\x65\x83\x61\x67\x65", 20);
// failure message
std::string fv1("Neo.ClientError.Statement.SyntaxError"), fv2("Invalid syntax.");
std::string fk1("code"), fk2("message");
TypedValue ftv1(fv1), ftv2(fv2);
std::map<std::string, TypedValue> fvals;
fvals.insert(std::make_pair(fk1, ftv1));
fvals.insert(std::make_pair(fk2, ftv2));
bolt_encoder.MessageFailure(fvals);
check_output(output, (const uint8_t*) "\xB1\x7F\xA2\x84\x63\x6F\x64\x65\xD0\x25\x4E\x65\x6F\x2E\x43\x6C\x69\x65\x6E\x74\x45\x72\x72\x6F\x72\x2E\x53\x74\x61\x74\x65\x6D\x65\x6E\x74\x2E\x53\x79\x6E\x74\x61\x78\x45\x72\x72\x6F\x72\x87\x6D\x65\x73\x73\x61\x67\x65\x8F\x49\x6E\x76\x61\x6C\x69\x64\x20\x73\x79\x6E\x74\x61\x78\x2E", 71);
// ignored message
bolt_encoder.MessageIgnored();
check_output(output, (const uint8_t*) "\xB0\x7E", 2);
}
int main(int argc, char** argv) {
initialize_data(data, SIZE);
logging::init_sync();
logging::log->pipe(std::make_unique<Stdout>());
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

50
tests/unit/bolt_result_stream.cpp Normal file
View File

@ -0,0 +1,50 @@
#include "bolt_common.hpp"
#include "communication/bolt/v1/encoder/chunked_buffer.hpp"
#include "communication/bolt/v1/encoder/encoder.hpp"
#include "communication/bolt/v1/encoder/result_stream.hpp"
#include "query/backend/cpp/typed_value.hpp"
using buffer_t = communication::bolt::ChunkedBuffer<TestSocket>;
using encoder_t = communication::bolt::Encoder<buffer_t, TestSocket>;
using result_stream_t = communication::bolt::ResultStream<TestSocket>;
const uint8_t header_output[] = "\x00\x29\xB1\x70\xA1\x86\x66\x69\x65\x6C\x64\x73\x9A\x82\x61\x61\x82\x62\x62\x82\x63\x63\x82\x64\x64\x82\x65\x65\x82\x66\x66\x82\x67\x67\x82\x68\x68\x82\x69\x69\x82\x6A\x6A\x00\x00";
const uint8_t result_output[] = "\x00\x0A\xB1\x71\x92\x05\x85\x68\x65\x6C\x6C\x6F\x00\x00";
const uint8_t summary_output[] = "\x00\x0C\xB1\x70\xA1\x87\x63\x68\x61\x6E\x67\x65\x64\x0A\x00\x00";
TEST(Bolt, ResultStream) {
TestSocket socket(10);
encoder_t encoder(socket);
result_stream_t result_stream(encoder);
std::vector<uint8_t>& output = socket.output;
std::vector<std::string> headers;
for (int i = 0; i < 10; ++i) headers.push_back(std::string(2, (char)('a' + i)));
result_stream.Header(headers);
print_output(output);
check_output(output, header_output, 45);
std::vector<TypedValue> result{TypedValue(5), TypedValue(std::string("hello"))};
result_stream.Result(result);
print_output(output);
check_output(output, result_output, 14);
std::map<std::string, TypedValue> summary;
summary.insert(std::make_pair(std::string("changed"), TypedValue(10)));
result_stream.Summary(summary);
print_output(output);
check_output(output, summary_output, 16);
}
int main(int argc, char** argv) {
logging::init_sync();
logging::log->pipe(std::make_unique<Stdout>());
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}

58
tests/unit/bolt_session.cpp
View File

@ -1,49 +1,9 @@
#include <array>
#include <cassert>
#include <cstring>
#include <iostream>
#include <vector>
#include "gtest/gtest.h"
#include "logging/streams/stdout.hpp"
#include "bolt_common.hpp"
#include "communication/bolt/v1/serialization/record_stream.hpp"
#include "communication/bolt/v1/session.hpp"
#include "dbms/dbms.hpp"
#include "query/engine.hpp"
class TestSocket {
public:
TestSocket(int socket) : socket(socket) {}
TestSocket(const TestSocket& s) : socket(s.id()){};
TestSocket(TestSocket&& other) { *this = std::forward<TestSocket>(other); }
TestSocket& operator=(TestSocket&& other) {
this->socket = other.socket;
other.socket = -1;
return *this;
}
void Close() { socket = -1; }
bool IsOpen() { return socket != -1; }
int id() const { return socket; }
int Write(const std::string& str) { return Write(str.c_str(), str.size()); }
int Write(const char* data, size_t len) {
return Write(reinterpret_cast<const uint8_t*>(data), len);
}
int Write(const uint8_t* data, size_t len) {
for (int i = 0; i < len; ++i) output.push_back(data[i]);
return len;
}
std::vector<uint8_t> output;
protected:
int socket;
};
const uint8_t handshake_req[] =
"\x60\x60\xb0\x17\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
@ -60,22 +20,6 @@ const uint8_t run_req[] =
"\x61\x6d\x65\x3a\x20\x32\x39\x33\x38\x33\x7d\x29\x20\x52\x45\x54\x55\x52"
"\x4e\x20\x6e\xa0\x00\x00";
void print_output(std::vector<uint8_t>& output) {
fprintf(stderr, "output: ");
for (int i = 0; i < output.size(); ++i) {
fprintf(stderr, "%02X ", output[i]);
}
fprintf(stderr, "\n");
}
void check_output(std::vector<uint8_t>& output, const uint8_t* data,
uint64_t len) {
EXPECT_EQ(len, output.size());
for (int i = 0; i < len; ++i) {
EXPECT_EQ(output[i], data[i]);
}
output.clear();
}
TEST(Bolt, Session) {
Dbms dbms;