diff --git a/include/communication/bolt/v1/serialization/bolt_serializer.hpp b/include/communication/bolt/v1/serialization/bolt_serializer.hpp index 73578a121..6abab5245 100644 --- a/include/communication/bolt/v1/serialization/bolt_serializer.hpp +++ b/include/communication/bolt/v1/serialization/bolt_serializer.hpp @@ -95,6 +95,16 @@ public: void write(const String &prop) { encoder.write_string(prop.value); } + void write_failure(const std::map& data) + { + encoder.message_failure(); + encoder.write_map_header(data.size()); + for (auto const &kv : data) { + write(kv.first); + write(kv.second); + } + } + template void handle(const T &prop) { diff --git a/include/communication/bolt/v1/serialization/record_stream.hpp b/include/communication/bolt/v1/serialization/record_stream.hpp index 93bd7c4e2..45b23683f 100644 --- a/include/communication/bolt/v1/serialization/record_stream.hpp +++ b/include/communication/bolt/v1/serialization/record_stream.hpp @@ -92,6 +92,11 @@ public: chunk(); } + void write_failure(const std::map& data) + { + serializer.write_failure(data); + chunk(); + } // -- BOLT SPECIFIC METHODS ----------------------------------------------- void write(const VertexAccessor &vertex) { serializer.write(vertex); } diff --git a/include/communication/bolt/v1/transport/bolt_encoder.hpp b/include/communication/bolt/v1/transport/bolt_encoder.hpp index 29a41207d..dba6ab9be 100644 --- a/include/communication/bolt/v1/transport/bolt_encoder.hpp +++ b/include/communication/bolt/v1/transport/bolt_encoder.hpp @@ -253,6 +253,12 @@ public: write(underlying_cast(MessageCode::Ignored)); } + void message_failure() + { + write_struct_header(1); + write(underlying_cast(MessageCode::Failure)); + } + void message_ignored_empty() { message_ignored(); diff --git a/src/communication/bolt/v1/states/executor.cpp b/src/communication/bolt/v1/states/executor.cpp index 589860ba5..b8e7e9777 100644 --- a/src/communication/bolt/v1/states/executor.cpp +++ b/src/communication/bolt/v1/states/executor.cpp @@ -10,7 +10,7 @@ namespace bolt Executor::Executor() : logger(logging::log->logger("Executor")) {} -State* Executor::run(Session& session) +State *Executor::run(Session &session) { // just read one byte that represents the struct type, we can skip the // information contained in this byte @@ -20,31 +20,22 @@ State* Executor::run(Session& session) auto message_type = session.decoder.read_byte(); - if(message_type == MessageCode::Run) - { + if (message_type == MessageCode::Run) { Query q; q.statement = session.decoder.read_string(); this->run(session, q); - } - else if(message_type == MessageCode::PullAll) - { + } else if (message_type == MessageCode::PullAll) { pull_all(session); - } - else if(message_type == MessageCode::DiscardAll) - { + } else if (message_type == MessageCode::DiscardAll) { discard_all(session); - } - else if(message_type == MessageCode::Reset) - { + } else if (message_type == MessageCode::Reset) { // todo rollback current transaction // discard all records waiting to be sent return this; - } - else - { + } else { logger.error("Unrecognized message recieved"); logger.debug("Invalid message type 0x{:02X}", message_type); @@ -54,7 +45,7 @@ State* Executor::run(Session& session) return this; } -void Executor::run(Session& session, Query& query) +void Executor::run(Session &session, Query &query) { logger.trace("[Run] '{}'", query.statement); @@ -64,18 +55,20 @@ void Executor::run(Session& session, Query& query) try { query_engine.execute(query.statement, db, session.output_stream); } catch (QueryEngineException &e) { - // return error to user + session.output_stream.write_failure( + {{"code", "unknown"}, {"message", e.what()}}); + session.output_stream.send(); } } -void Executor::pull_all(Session& session) +void Executor::pull_all(Session &session) { logger.trace("[PullAll]"); session.output_stream.send(); } -void Executor::discard_all(Session& session) +void Executor::discard_all(Session &session) { logger.trace("[DiscardAll]"); @@ -85,5 +78,4 @@ void Executor::discard_all(Session& session) session.output_stream.chunk(); session.output_stream.send(); } - }