Merge pull request #32 in ASAPO/asapo from feature_worker-perf to develop

* commit 'f7b51f60': (47 commits)
  refactor
  using epool for linux data server
  mofified worker example
  release connection when no data found, update tests, correct file permissions
  update nomad config
  fix test
  fix mem leaks
  add statistics for data server
  refactoring statistics class
  fix windows tests
  fix windows tests
  fix windows tests
  refactoring, more intergation tests
  fix windows tests
  fix windows tests
  add integration test for memory buffer
  finished client
  bugfix
  started working at client for memory buffer
  server part of mamory cache
  ...

CMakeLists.txt

@@ -3,6 +3,7 @@ project(ASAPO)
 set(CMAKE_CXX_STANDARD 11)
 IF(WIN32)
     set(CMAKE_CXX_FLAGS_DEBUG "/MTd")
+    set(CMAKE_CXX_FLAGS_RELEASE "/MT")
     add_definitions(-DWIN32)
 ELSEIF(CMAKE_C_COMPILER_ID STREQUAL "GNU")
     SET( CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} -static-libgcc -static-libstdc++")
@@ -47,11 +48,12 @@ message (STATUS "cURL include: ${CURL_INCLUDE_DIRS}")
 # format sources
 include(astyle)
 
+include(prepare_version_tag)
+
 include(testing_cpp)
 
 include(prepare_asapo)
 
-
 if(BUILD_WORKER_TOOLS)
     set (BUILD_MONGODB_CLIENTLIB ON)
 endif()

CMakeModules/prepare_asapo.cmake

@@ -5,12 +5,22 @@ function(prepare_asapo)
     get_target_property(AUTHORIZER_FULLPATH asapo-authorizer EXENAME)
     get_target_property(BROKER_FULLPATH asapo-broker EXENAME)
     set(WORK_DIR ${CMAKE_CURRENT_BINARY_DIR})
+
+    if(NOT DEFINED RECEIVER_USE_CACHE)
+        set(RECEIVER_USE_CACHE true)
+    endif()
+
+    if(NOT DEFINED RECEIVER_WRITE_TO_DISK)
+        set(RECEIVER_WRITE_TO_DISK true)
+    endif()
+
     if (WIN32)
-        configure_file(${CMAKE_SOURCE_DIR}/tests/automatic/settings/receiver.json.tpl.win receiver.json.tpl COPYONLY)
+        configure_file(${CMAKE_SOURCE_DIR}/tests/automatic/settings/receiver.json.tpl.win.in receiver.json.tpl @ONLY)
     else()
-        configure_file(${CMAKE_SOURCE_DIR}/tests/automatic/settings/receiver.json.tpl.lin receiver.json.tpl COPYONLY)
+        configure_file(${CMAKE_SOURCE_DIR}/tests/automatic/settings/receiver.json.tpl.lin.in receiver.json.tpl @ONLY)
     endif()
-    configure_file(${CMAKE_SOURCE_DIR}/config/nomad/receiver.nmd.in  receiver.nmd @ONLY)
+
+      configure_file(${CMAKE_SOURCE_DIR}/config/nomad/receiver.nmd.in  receiver.nmd @ONLY)
     configure_file(${CMAKE_SOURCE_DIR}/config/nomad/discovery.nmd.in  discovery.nmd @ONLY)
     configure_file(${CMAKE_SOURCE_DIR}/config/nomad/authorizer.nmd.in  authorizer.nmd @ONLY)
     configure_file(${CMAKE_SOURCE_DIR}/config/nomad/broker.nmd.in  broker.nmd @ONLY)

CMakeModules/prepare_version.cmake

 execute_process(COMMAND git describe --tags --dirty OUTPUT_VARIABLE VERSION)
 string(STRIP ${VERSION} VERSION)
 
+
+execute_process(COMMAND git rev-parse --abbrev-ref HEAD OUTPUT_VARIABLE BRANCH)
+string(STRIP ${BRANCH} BRANCH)
+
+if (${BRANCH} STREQUAL "develop")
+    SET (ASAPO_VERSION ${BRANCH}.${VERSION})
+else()
+    SET (ASAPO_VERSION ${BRANCH}.latest)
+endif()
+
 string(TIMESTAMP TIMESTAMP "%H:%M:%S %d.%m.%Y UTC" UTC)
 
 configure_file(${PROJECT_SOURCE_DIR}/common/cpp/include/common/version.h.in ${PROJECT_SOURCE_DIR}/common/cpp/include/common/version.h @ONLY)

CMakeModules/prepare_version_tag.cmake

+execute_process(COMMAND git describe --tags --dirty OUTPUT_VARIABLE VERSION)
+string(STRIP ${VERSION} VERSION)
+
+execute_process(COMMAND git rev-parse --abbrev-ref HEAD OUTPUT_VARIABLE BRANCH)
+string(STRIP ${BRANCH} BRANCH)
+
+if (${BRANCH} STREQUAL "develop")
+    SET (ASAPO_VERSION ${BRANCH}.${VERSION})
+else()
+    SET (ASAPO_VERSION ${BRANCH}.latest)
+endif()
+

common/cpp/CMakeLists.txt

@@ -10,6 +10,7 @@ add_subdirectory(src/http_client)
 
 add_subdirectory(src/logger)
 
+add_subdirectory(src/request)
 
 if(BUILD_MONGODB_CLIENTLIB)
     add_subdirectory(src/database)

common/cpp/include/common/data_structs.h

@@ -15,9 +15,11 @@ class FileInfo {
     std::chrono::system_clock::time_point modify_date;
     uint64_t size{0};
     uint64_t id{0};
+    std::string source;
+    uint64_t buf_id{0};
     std::string Json() const;
     bool SetFromJson(const std::string& json_string);
-    std::string FullName(const std::string& base_path);
+    std::string FullName(const std::string& base_path) const;
 };
 
 inline bool operator==(const FileInfo& lhs, const FileInfo& rhs) {

common/cpp/include/common/error.h

@@ -10,7 +10,7 @@ namespace asapo {
 enum class ErrorType {
     kUnknownError,
 
-    kHidraError,
+    kAsapoError,
     kHttpError,
     kIOError,
     kReceiverError,
@@ -46,8 +46,8 @@ class ErrorTemplateInterface {
   public:
     virtual ErrorType GetErrorType() const noexcept = 0;
     virtual Error Generate() const noexcept = 0;
+    virtual Error Generate(const std::string& suffix) const noexcept = 0;
     virtual std::string Text() const noexcept = 0;
-
     virtual inline bool operator == (const Error& rhs) const {
         return rhs != nullptr &&
                GetErrorType() == rhs->GetErrorType();
@@ -79,7 +79,7 @@ static inline std::ostream& operator<<(std::ostream& os, const Error& err) {
 class SimpleError: public ErrorInterface {
   private:
     std::string error_;
-    ErrorType error_type_ = ErrorType::kHidraError;
+    ErrorType error_type_ = ErrorType::kAsapoError;
   public:
     explicit SimpleError(std::string error): error_{std::move(error)} {
 
@@ -109,7 +109,7 @@ class SimpleError: public ErrorInterface {
 class SimpleErrorTemplate : public ErrorTemplateInterface {
   protected:
     std::string error_;
-    ErrorType error_type_ = ErrorType::kHidraError;
+    ErrorType error_type_ = ErrorType::kAsapoError;
   public:
     explicit SimpleErrorTemplate(std::string error): error_{std::move(error)} {
 
@@ -130,6 +130,11 @@ class SimpleErrorTemplate : public ErrorTemplateInterface {
     inline Error Generate() const noexcept override {
         return Error(new SimpleError{error_, error_type_});
     }
+
+    inline Error Generate(const std::string& suffix) const noexcept override {
+        return Error(new SimpleError{error_ + " :" + suffix, error_type_});
+    }
+
 };
 
 static inline std::ostream& operator<<(std::ostream& os, const SimpleErrorTemplate& err) {
@@ -155,5 +160,47 @@ auto const kEndOfFile = SimpleErrorTemplate {
 
 }
 
+template <typename ServiceErrorType, ErrorType MainErrorType>
+class ServiceError : public SimpleError {
+  private:
+    ServiceErrorType error_type_;
+  public:
+    ServiceError(const std::string& error, ServiceErrorType error_type) : SimpleError(error, MainErrorType) {
+        error_type_ = error_type;
+    }
+    ServiceErrorType GetServiceErrorType() const noexcept {
+        return error_type_;
+    }
+};
+
+template <typename ServiceErrorType, ErrorType MainErrorType>
+class ServiceErrorTemplate : public SimpleErrorTemplate {
+  protected:
+    ServiceErrorType error_type_;
+  public:
+    ServiceErrorTemplate(const std::string& error, ServiceErrorType error_type) : SimpleErrorTemplate(error,
+                MainErrorType) {
+        error_type_ = error_type;
+    }
+
+    inline ServiceErrorType GetServiceErrorType() const noexcept {
+        return error_type_;
+    }
+
+    inline Error Generate() const noexcept override {
+        auto err = new ServiceError<ServiceErrorType, MainErrorType>(error_, error_type_);
+        return Error(err);
+    }
+
+    inline Error Generate(const std::string& suffix) const noexcept override {
+        return Error(new ServiceError<ServiceErrorType, MainErrorType>(error_ + " :" + suffix, error_type_));
+    }
+
+    inline bool operator==(const Error& rhs) const override {
+        return SimpleErrorTemplate::operator==(rhs)
+               && GetServiceErrorType() == ((ServiceError<ServiceErrorType, MainErrorType>*) rhs.get())->GetServiceErrorType();
+    }
+};
+
 }
 #endif //ASAPO_ERROR_H

common/cpp/include/common/io_error.h

@@ -30,45 +30,8 @@ enum class IOErrorType {
 
 };
 
-class IOError : public SimpleError {
-  private:
-    IOErrorType io_error_type_;
-  public:
-    IOError(const std::string& error, IOErrorType io_error_type) : SimpleError(error, ErrorType::kIOError) {
-        io_error_type_ = io_error_type;
-    }
-
-    IOErrorType GetIOErrorType() const noexcept {
-        return io_error_type_;
-    }
-};
-
-class IOErrorTemplate : public SimpleErrorTemplate {
-  protected:
-    IOErrorType io_error_type_;
-  public:
-    IOErrorTemplate(const std::string& error, IOErrorType io_error_type) : SimpleErrorTemplate(error, ErrorType::kIOError) {
-        io_error_type_ = io_error_type;
-    }
-
-    inline IOErrorType GetIOErrorType() const noexcept {
-        return io_error_type_;
-    }
-
-    inline Error Generate() const noexcept override {
-        return Error(new IOError(error_, io_error_type_));
-    }
-
-    inline bool operator == (const Error& rhs) const override {
-        return SimpleErrorTemplate::operator==(rhs)
-               && GetIOErrorType() == ((IOError*)rhs.get())->GetIOErrorType();
-    }
-};
-
-static inline std::ostream& operator<<(std::ostream& os, const IOErrorTemplate& err) {
-    return os << err.Text();
-}
-
+using IOError = ServiceError<IOErrorType, ErrorType::kIOError>;
+using IOErrorTemplate = ServiceErrorTemplate<IOErrorType, ErrorType::kIOError>;
 
 namespace IOErrorTemplates {
 auto const kUnknownIOError = IOErrorTemplate {

common/cpp/include/common/networking.h

@@ -13,12 +13,15 @@ typedef uint64_t NetworkRequestId;
 enum Opcode : uint8_t {
     kOpcodeUnknownOp = 1,
     kOpcodeTransferData,
+    kOpcodeGetBufferData,
     kOpcodeAuthorize,
     kOpcodeCount,
 };
 
 enum NetworkErrorCode : uint16_t {
     kNetErrorNoError,
+    kNetErrorWrongRequest,
+    kNetErrorNoData,
     kNetAuthorizationError,
     kNetErrorFileIdAlreadyInUse,
     kNetErrorAllocateStorageFailed,
@@ -35,6 +38,11 @@ struct GenericRequestHeader {
         op_code{i_op_code}, data_id{i_data_id}, data_size{i_data_size} {
         strncpy(message, i_message.c_str(), kMaxMessageSize);
     }
+    GenericRequestHeader(const GenericRequestHeader& header) {
+        op_code = header.op_code, data_id = header.data_id, data_size = header.data_size,
+        strncpy(message, header.message, kMaxMessageSize);
+    }
+
     Opcode      op_code;
     uint64_t    data_id;
     uint64_t    data_size;
@@ -43,7 +51,6 @@ struct GenericRequestHeader {
 
 struct GenericNetworkResponse {
     Opcode              op_code;
-    NetworkRequestId    request_id;
     NetworkErrorCode    error_code;
     char        message[kMaxMessageSize];
 };

common/cpp/include/io/io.h

@@ -41,6 +41,7 @@ enum class SocketProtocols {
 
 using FileDescriptor = int;
 using SocketDescriptor = int;
+using ListSocketDescriptors =  std::vector<SocketDescriptor>;
 const SocketDescriptor kDisconnectedSocketDescriptor = -1;
 
 class IO {
@@ -57,6 +58,12 @@ class IO {
     virtual SocketDescriptor  CreateSocket(AddressFamilies address_family, SocketTypes socket_type,
                                            SocketProtocols socket_protocol, Error* err) const = 0;
     virtual void            Listen(SocketDescriptor socket_fd, int backlog, Error* err) const = 0;
+
+    virtual ListSocketDescriptors WaitSocketsActivity(SocketDescriptor master_socket,
+                                                      ListSocketDescriptors* sockets_to_listen,
+                                                      std::vector<std::string>* new_connections, Error* err) const = 0;
+
+
     virtual void            InetBind(SocketDescriptor socket_fd, const std::string& address, Error* err) const = 0;
     virtual SocketDescriptor  CreateAndBindIPTCPSocketListener(const std::string& address, int backlog,
             Error* err) const = 0;
@@ -102,6 +109,9 @@ class IO {
     virtual std::vector<FileInfo>   FilesInFolder   (const std::string& folder, Error* err) const = 0;
     virtual std::string     ReadFileToString        (const std::string& fname, Error* err) const = 0;
     virtual Error GetLastError() const = 0;
+    virtual std::string AddressFromSocket(SocketDescriptor socket) const noexcept = 0;
+    virtual std::string     GetHostName(Error* err) const noexcept = 0;
+    virtual ~IO() = default;
 };
 
 }

common/cpp/include/request/request.h

+#ifndef ASAPO_GENERIC_REQUEST_H
+#define ASAPO_GENERIC_REQUEST_H
+
+#include "common/networking.h"
+#include "common/data_structs.h"
+#include "io/io.h"
+
+namespace asapo {
+
+
+class GenericRequest {
+  public:
+    GenericRequest() = delete;
+    GenericRequest(GenericRequestHeader h): header{std::move(h)} {};
+    GenericRequestHeader header;
+    virtual ~GenericRequest() = default;
+};
+
+using GenericRequestPtr = std::unique_ptr<GenericRequest>;
+using GenericRequests = std::vector<GenericRequestPtr>;
+
+}
+
+#endif //ASAPO_GENERIC_REQUEST_H

producer/api/src/request_handler.h → common/cpp/include/request/request_handler.h

-#ifndef ASAPO_PRODUCER_REQUEST_HANDLER_H
-#define ASAPO_PRODUCER_REQUEST_HANDLER_H
+#ifndef ASAPO_REQUEST_HANDLER_H
+#define ASAPO_REQUEST_HANDLER_H
 
 #include <memory>
 
 #include "common/error.h"
+
 #include "request.h"
 
 namespace  asapo {
@@ -12,11 +13,11 @@ class RequestHandler {
   public:
     virtual void PrepareProcessingRequestLocked()  = 0;
     virtual void TearDownProcessingRequestLocked(const Error& error_from_process)  = 0;
-    virtual Error ProcessRequestUnlocked(Request* request)  = 0;
+    virtual Error ProcessRequestUnlocked(GenericRequest* request)  = 0;
     virtual bool ReadyProcessRequest() = 0;
     virtual ~RequestHandler() = default;
 };
 
 
 }
-#endif //ASAPO_PRODUCER_REQUEST_HANDLER_H
+#endif //ASAPO_REQUEST_HANDLER_H

common/cpp/include/request/request_handler_factory.h

+#ifndef ASAPO_REQUEST_HANDLER_FACTORY_H
+#define ASAPO_REQUEST_HANDLER_FACTORY_H
+
+#include <memory>
+
+#include "request_handler.h"
+
+namespace  asapo {
+
+class RequestHandlerFactory {
+  public:
+    virtual std::unique_ptr<RequestHandler> NewRequestHandler(uint64_t thread_id, uint64_t* shared_counter) = 0;
+};
+
+
+}
+
+#endif //ASAPO_REQUEST_HANDLER_FACTORY_H

producer/api/src/request_pool.h → common/cpp/include/request/request_pool.h

@@ -8,11 +8,9 @@
 #include <condition_variable>
 #include <queue>
 
-
 #include "logger/logger.h"
-#include "request_handler_tcp.h"
 #include "request_handler_factory.h"
-
+#include "request.h"
 #include "preprocessor/definitions.h"
 
 
@@ -23,23 +21,25 @@ class RequestPool {
         std::unique_lock<std::mutex> lock;
     };
   public:
-    explicit RequestPool(uint8_t n_threads, RequestHandlerFactory* request_handler_factory);
-    VIRTUAL Error AddRequest(std::unique_ptr<Request> request);
+    explicit RequestPool(uint8_t n_threads, RequestHandlerFactory* request_handler_factory, const AbstractLogger* log);
+    VIRTUAL Error AddRequest(GenericRequestPtr request);
+    VIRTUAL Error AddRequests(GenericRequests requests);
+
     ~RequestPool();
-    AbstractLogger* log__;
     uint64_t NRequestsInQueue();
   private:
+    const AbstractLogger* log__;
     RequestHandlerFactory* request_handler_factory__;
     std::vector<std::thread> threads_;
     void ThreadHandler(uint64_t id);
     bool quit_{false};
     std::condition_variable condition_;
     std::mutex mutex_;
-    std::deque<std::unique_ptr<Request>> request_queue_;
+    std::deque<GenericRequestPtr> request_queue_;
     bool CanProcessRequest(const std::unique_ptr<RequestHandler>& request_handler);
     void ProcessRequest(const std::unique_ptr<RequestHandler>& request_handler, ThreadInformation* thread_info);
-    std::unique_ptr<Request> GetRequestFromQueue();
-    void PutRequestBackToQueue(std::unique_ptr<Request>request);
+    GenericRequestPtr GetRequestFromQueue();
+    void PutRequestBackToQueue(GenericRequestPtr request);
     uint64_t shared_counter_{0};
 
 };

common/cpp/include/unittests/MockIO.h

@@ -8,11 +8,42 @@
 namespace asapo {
 class MockIO : public IO {
   public:
+
+    std::string GetHostName(Error* err) const noexcept override {
+        ErrorInterface* error = nullptr;
+        auto res = GetHostName_t(&error);
+        err->reset(error);
+        return res;
+
+    }
+    MOCK_CONST_METHOD1(GetHostName_t, std::string(ErrorInterface** err));
+
+
+    std::string AddressFromSocket(SocketDescriptor socket) const noexcept override {
+        return AddressFromSocket_t(socket);
+    }
+    MOCK_CONST_METHOD1(AddressFromSocket_t, std::string (SocketDescriptor socket));
+
+
     std::unique_ptr<std::thread> NewThread(std::function<void()> function) const override {
         return std::unique_ptr<std::thread>(NewThread_t(function));
     }
     MOCK_CONST_METHOD1(NewThread_t, std::thread * (std::function<void()> function));
 
+
+    ListSocketDescriptors WaitSocketsActivity(SocketDescriptor master_socket, ListSocketDescriptors* sockets_to_listen,
+                                              std::vector<std::string>* new_connections, Error* err) const override {
+        ErrorInterface* error = nullptr;
+        auto data = WaitSocketsActivity_t(master_socket, sockets_to_listen, new_connections, &error);
+        err->reset(error);
+        return data;
+    }
+
+    MOCK_CONST_METHOD4(WaitSocketsActivity_t, ListSocketDescriptors(SocketDescriptor master_socket,
+                       ListSocketDescriptors* sockets_to_listen,
+                       std::vector<std::string>* connections, ErrorInterface** err));
+
+
     SocketDescriptor CreateSocket(AddressFamilies address_family, SocketTypes socket_type, SocketProtocols socket_protocol,
                                   Error* err) const override {
         ErrorInterface* error = nullptr;

common/cpp/src/data_structs/data_structs.cpp

@@ -2,20 +2,27 @@
 
 #include "json_parser/json_parser.h"
 
+#include "preprocessor/definitions.h"
+
+#include <iostream>
+
 namespace asapo {
 
 
 std::string FileInfo::Json() const {
     auto nanoseconds_from_epoch = std::chrono::time_point_cast<std::chrono::nanoseconds>(modify_date).
                                   time_since_epoch().count();
+    int64_t buf_id_int = static_cast<int64_t>(buf_id);
     std::string s = "{\"_id\":" + std::to_string(id) + ","
                     "\"size\":" + std::to_string(size) + ","
                     "\"name\":\"" + name + "\","
-                    "\"lastchange\":" + std::to_string(nanoseconds_from_epoch) + "}";
+                    "\"lastchange\":" + std::to_string(nanoseconds_from_epoch) + ","
+                    "\"source\":\"" + source + "\","
+                    "\"buf_id\":" + std::to_string(buf_id_int)
+                    + "}";
     return s;
 }
 
-
 bool TimeFromJson(const JsonStringParser& parser, const std::string name, std::chrono::system_clock::time_point* val) {
     uint64_t nanoseconds_from_epoch;
     if (parser.GetUInt64(name, &nanoseconds_from_epoch)) {
@@ -39,6 +46,8 @@ bool FileInfo::SetFromJson(const std::string& json_string) {
     if (parser.GetUInt64("_id", &id) ||
             parser.GetUInt64("size", &size) ||
             parser.GetString("name", &name) ||
+            parser.GetString("source", &source) ||
+            parser.GetUInt64("buf_id", &buf_id) ||
             !TimeFromJson(parser, "lastchange", &modify_date)) {
         *this = old;
         return false;
@@ -47,9 +56,9 @@ bool FileInfo::SetFromJson(const std::string& json_string) {
     return true;
 }
 
-std::string FileInfo::FullName(const std::string& base_path) {
+std::string FileInfo::FullName(const std::string& base_path) const  {
     std::string full_name;
-    full_name = base_path.empty() ? "" : base_path + "/";
+    full_name = base_path.empty() ? "" : base_path + kPathSeparator;
     return full_name + name;
 }
 

common/cpp/src/database/mongodb_client.cpp

@@ -93,11 +93,12 @@ void MongoDBClient::CleanUp() {
 }
 
 bson_p PrepareBsonDocument(const FileInfo& file, Error* err) {
+    bson_error_t mongo_err;
     auto s = file.Json();
     auto json = reinterpret_cast<const uint8_t*>(s.c_str());
-    auto bson = bson_new_from_json(json, -1, nullptr);
+    auto bson = bson_new_from_json(json, -1, &mongo_err);
     if (!bson) {
-        *err = TextError(DBError::kInsertError);
+        *err = TextError(std::string(DBError::kInsertError) + ": " + mongo_err.message);
         return nullptr;
     }
 

common/cpp/src/http_client/curl_http_client.cpp

@@ -108,7 +108,6 @@ CurlHttpClient::~CurlHttpClient() {
     if (curl_) {
         curl_easy_cleanup(curl_);
     }
-
 }
 
 

common/cpp/src/json_parser/rapid_json.cpp

@@ -3,13 +3,15 @@
 
 using namespace rapidjson;
 
+#include <iostream>
+
 namespace asapo {
 
-RapidJson::RapidJson(const std::string& json, const std::unique_ptr<IO>* io): io__{io}, json_{json} {
+RapidJson::RapidJson(const std::string& json, const std::unique_ptr<IO>* io) : io__{io}, json_{json} {
 
 }
 
-Error RapidJson::LazyInitialize()const noexcept {
+Error RapidJson::LazyInitialize() const noexcept {
     if (embedded_error_) {
         return TextError(embedded_error_->Explain());
     }
@@ -26,7 +28,7 @@ Error RapidJson::LazyInitialize()const noexcept {
         }
     }
 
-    if ( doc_.Parse(str.c_str()).HasParseError()) {
+    if (doc_.Parse(str.c_str()).HasParseError()) {
         return TextError("Cannot parse document");
     }
 
@@ -46,6 +48,9 @@ asapo::Error RapidJson::CheckValueType(const std::string& name, ValueType type,
         res = val->IsString();
         break;
     case ValueType::kUint64:
+        res = val->IsUint64();
+        break;
+    case ValueType::kInt64:
         res = val->IsInt64();
         break;
     case ValueType::kBool:
@@ -62,27 +67,46 @@ asapo::Error RapidJson::CheckValueType(const std::string& name, ValueType type,
     return nullptr;
 }
 
-
-asapo::Error RapidJson::GetValuePointer(const std::string& name, ValueType type, Value** val)const noexcept {
+asapo::Error RapidJson::GetValuePointer(const std::string& name, ValueType type, Value** val) const noexcept {
     if (Error err = LazyInitialize()) {
         return err;
     }
 
     auto iterator = object_p_->FindMember(name.c_str());
     if (iterator == object_p_->MemberEnd()) {
-        return  TextError("cannot find: " + name);
+        return TextError("cannot find: " + name);
     }
 
-    *val =  &iterator->value;
+    *val = &iterator->value;
     return CheckValueType(name, type, *val);
 }
 
+Error RapidJson::GetInt64(const std::string& name, int64_t* val) const noexcept {
+    Value* json_val;
+    if (Error err = GetValuePointer(name, ValueType::kInt64, &json_val)) {
+        return err;
+    }
+    *val = json_val->GetInt64();
+    return nullptr;
+}
+
 Error RapidJson::GetUInt64(const std::string& name, uint64_t* val) const noexcept {
+    int64_t val_int64;
+
+    Error err = GetInt64(name, &val_int64);
+    if (!initialized_) {
+        return err;
+    }
+    if (err == nullptr) {
+        *val = static_cast<uint64_t>(val_int64);
+        return nullptr;
+    }
+
     Value* json_val;
     if (Error err = GetValuePointer(name, ValueType::kUint64, &json_val)) {
         return err;
     }
-    *val = json_val->GetInt64();
+    *val = json_val->GetUint64();
     return nullptr;
 }
 
@@ -104,7 +128,6 @@ Error RapidJson::GetString(const std::string& name, std::string* val) const noex
     return nullptr;
 }
 
-
 Error RapidJson::GetArrayUInt64(const std::string& name, std::vector<uint64_t>* val) const noexcept {
     Value* json_val;
     if (Error err = GetValuePointer(name, ValueType::kArray, &json_val)) {
@@ -113,10 +136,10 @@ Error RapidJson::GetArrayUInt64(const std::string& name, std::vector<uint64_t>*
 
     val->clear();
     for (auto& v : json_val->GetArray()) {
-        if (!v.IsInt64()) {
+        if (!v.IsUint64()) {
             return TextError("wrong type of array element: " + name);
         }
-        val->push_back(v.GetInt());
+        val->push_back(v.GetUint64());
     }
     return nullptr;
 
@@ -148,5 +171,4 @@ RapidJson::RapidJson(const RapidJson& parent, const std::string& subname) {
     initialized_ = true;
 }
 
-
 }
\ No newline at end of file

common/cpp/src/json_parser/rapid_json.h

@@ -10,6 +10,7 @@ namespace asapo {
 
 enum class ValueType {
     kUint64,
+    kInt64,
     kString,
     kObject,
     kArray,
@@ -26,6 +27,7 @@ class RapidJson {
     Error GetArrayUInt64(const std::string& name, std::vector<uint64_t>* val) const noexcept;
     Error GetArrayString(const std::string& name, std::vector<std::string>* val) const noexcept;
   private:
+    Error GetInt64(const std::string& name, int64_t* val) const noexcept;
     const std::unique_ptr<IO>* io__;
     mutable rapidjson::Document doc_;
     mutable rapidjson::Value object_;