From c98e69e3dc2cf4ec7af5d397d23b9668cdcab315 Mon Sep 17 00:00:00 2001
From: Sergey Yakubov <sergey.yakubov@desy.de>
Date: Fri, 21 Jun 2019 15:55:38 +0200
Subject: [PATCH] started working at broker
---
worker/api/cpp/include/worker/data_broker.h | 9 ++
worker/api/cpp/src/folder_data_broker.cpp | 5 ++
worker/api/cpp/src/folder_data_broker.h | 1 +
worker/api/cpp/src/server_data_broker.cpp | 82 ++++++++++++++-----
worker/api/cpp/src/server_data_broker.h | 20 ++++-
.../api/cpp/unittests/test_folder_broker.cpp | 10 +++
.../api/cpp/unittests/test_server_broker.cpp | 22 +++++
7 files changed, 123 insertions(+), 26 deletions(-)
diff --git a/worker/api/cpp/include/worker/data_broker.h b/worker/api/cpp/include/worker/data_broker.h
index fbd5da3eb..c53bbacf8 100644
--- a/worker/api/cpp/include/worker/data_broker.h
+++ b/worker/api/cpp/include/worker/data_broker.h
@@ -91,6 +91,15 @@ class DataBroker {
\return Error if both pointers are nullptr or data cannot be read, nullptr otherwise.
*/
virtual Error GetLast(FileInfo* info, std::string group_id, FileData* data) = 0;
+
+ //! Get all images matching the query.
+ /*!
+ \param sql_query - query string in SQL format. Limit subset is supported
+ \param err - will be set in case of error, nullptr otherwise
+ \return vector of image metadata matchiing to specified query. Empty if nothing found or error
+ */
+ virtual FileInfos QueryImages(std::string query,Error* err) = 0;
+
virtual ~DataBroker() = default; // needed for unique_ptr to delete itself
};
diff --git a/worker/api/cpp/src/folder_data_broker.cpp b/worker/api/cpp/src/folder_data_broker.cpp
index f5eab447d..ed044b951 100644
--- a/worker/api/cpp/src/folder_data_broker.cpp
+++ b/worker/api/cpp/src/folder_data_broker.cpp
@@ -98,4 +98,9 @@ std::string FolderDataBroker::GetBeamtimeMeta(Error* err) {
return io__->ReadFileToString(base_path_ + kPathSeparator + "beamtime_global.meta", err);
}
+FileInfos FolderDataBroker::QueryImages(std::string query, Error* err) {
+ *err=TextError("Not supported for folder data broker");
+ return FileInfos{};
+}
+
}
diff --git a/worker/api/cpp/src/folder_data_broker.h b/worker/api/cpp/src/folder_data_broker.h
index 54319bbcc..aa0dfca5d 100644
--- a/worker/api/cpp/src/folder_data_broker.h
+++ b/worker/api/cpp/src/folder_data_broker.h
@@ -24,6 +24,7 @@ class FolderDataBroker final : public asapo::DataBroker {
uint64_t GetNDataSets(Error* err) override;
Error GetById(uint64_t id, FileInfo* info, std::string group_id, FileData* data) override;
std::unique_ptr<asapo::IO> io__; // modified in testings to mock system calls,otherwise do not touch
+ FileInfos QueryImages(std::string query,Error* err) override;
private:
std::string base_path_;
bool is_connected_;
diff --git a/worker/api/cpp/src/server_data_broker.cpp b/worker/api/cpp/src/server_data_broker.cpp
index 885e54c1f..a1491a53e 100644
--- a/worker/api/cpp/src/server_data_broker.cpp
+++ b/worker/api/cpp/src/server_data_broker.cpp
@@ -88,14 +88,13 @@ std::string ServerDataBroker::RequestWithToken(std::string uri) {
return std::move(uri) + "?token=" + token_;
}
-Error ServerDataBroker::ProcessRequest(std::string* response, std::string request_uri, std::string extra_params,
- bool post) {
+Error ServerDataBroker::ProcessRequest(std::string* response, const RequestInfo& request) {
Error err;
HttpCode code;
- if (post) {
- *response = httpclient__->Post(RequestWithToken(request_uri) + extra_params, "", &code, &err);
+ if (request.post) {
+ *response = httpclient__->Post(RequestWithToken(request.host+request.api) + request.extra_params, request.body, &code, &err);
} else {
- *response = httpclient__->Get(RequestWithToken(request_uri) + extra_params, &code, &err);
+ *response = httpclient__->Get(RequestWithToken(request.host+request.api) + request.extra_params, &code, &err);
}
if (err != nullptr) {
current_broker_uri_ = "";
@@ -109,9 +108,12 @@ Error ServerDataBroker::GetBrokerUri() {
return nullptr;
}
- std::string request_uri = server_uri_ + "/discovery/broker";
+ RequestInfo ri;
+ ri.host = server_uri_;
+ ri.api= "/discovery/broker";
+
Error err;
- err = ProcessRequest(¤t_broker_uri_, request_uri, "", false);
+ err = ProcessRequest(¤t_broker_uri_, ri);
if (err != nullptr || current_broker_uri_.empty()) {
current_broker_uri_ = "";
return TextError("cannot get broker uri from " + server_uri_);
@@ -128,7 +130,10 @@ Error ServerDataBroker::GetFileInfoFromServer(FileInfo* info, std::string group_
while (true) {
auto err = GetBrokerUri();
if (err == nullptr) {
- err = ProcessRequest(&response, current_broker_uri_ + request_api + request_suffix, "", false);
+ RequestInfo ri;
+ ri.host = current_broker_uri_;
+ ri.api = request_api + request_suffix;
+ err = ProcessRequest(&response, ri);
if (err == nullptr) {
break;
}
@@ -218,18 +223,29 @@ Error ServerDataBroker::TryGetDataFromBuffer(const FileInfo* info, FileData* dat
std::string ServerDataBroker::GenerateNewGroupId(Error* err) {
- return BrokerRequestWithTimeout("creategroup", "", true, err);
+ RequestInfo ri;
+ ri.api = "/creategroup";
+ ri.post = true;
+ return BrokerRequestWithTimeout(ri, err);
+}
+
+
+std::string ServerDataBroker::AppendUri(std::string request_string) {
+ return current_broker_uri_ + "/"+std::move(request_string);
}
-std::string ServerDataBroker::BrokerRequestWithTimeout(std::string request_string, std::string extra_params,
- bool post_request, Error* err) {
+
+
+std::string ServerDataBroker::BrokerRequestWithTimeout(RequestInfo request, Error* err) {
uint64_t elapsed_ms = 0;
std::string response;
while (elapsed_ms <= timeout_ms_) {
*err = GetBrokerUri();
if (*err == nullptr) {
- *err = ProcessRequest(&response, current_broker_uri_ + "/" + request_string, extra_params, post_request);
- if (*err == nullptr || (*err)->GetErrorType() == ErrorType::kEndOfFile) {
+ request.host = current_broker_uri_;
+ *err = ProcessRequest(&response, request);
+ if (*err == nullptr || (*err)->GetErrorType() == ErrorType::kEndOfFile ){
+// || (*err) == kWrongInput) {
return response;
}
}
@@ -241,18 +257,23 @@ std::string ServerDataBroker::BrokerRequestWithTimeout(std::string request_strin
}
Error ServerDataBroker::ResetCounter(std::string group_id) {
- std::string request_string = "database/" + source_name_ + "/" + std::move(group_id) + "/resetcounter";
+ RequestInfo ri;
+ ri.api = "/database/" + source_name_ + "/" + std::move(group_id) + "/resetcounter";
+ ri.post = true;
+
Error err;
- BrokerRequestWithTimeout(request_string, "", true, &err);
+ BrokerRequestWithTimeout(ri, &err);
return err;
}
uint64_t ServerDataBroker::GetNDataSets(Error* err) {
- std::string request_string = "database/" + source_name_ + "/size";
- auto responce = BrokerRequestWithTimeout(request_string, "", false, err);
+ RequestInfo ri;
+ ri.api = "/database/" + source_name_ + "/size";
+ auto responce = BrokerRequestWithTimeout(ri, err);
if (*err) {
return 0;
}
+
JsonStringParser parser(responce);
uint64_t size;
if ((*err = parser.GetUInt64("size", &size)) != nullptr) {
@@ -267,10 +288,14 @@ Error ServerDataBroker::GetById(uint64_t id, FileInfo* info, std::string group_i
Error ServerDataBroker::GetFileInfoFromServerById(uint64_t id, FileInfo* info, std::string group_id) {
- std::string request_string = "database/" + source_name_ + "/" + std::move(group_id) + "/" + std::to_string(id);
- std::string extra_params = "&reset=true";
+
+ RequestInfo ri;
+ ri.api = "/database/" + source_name_ + "/" + std::move(group_id) + "/" + std::to_string(id);
+ ri.extra_params = "&reset=true";
+
+
Error err;
- auto responce = BrokerRequestWithTimeout(request_string, extra_params, false, &err);
+ auto responce = BrokerRequestWithTimeout(ri, &err);
if (err) {
return err;
}
@@ -283,8 +308,21 @@ Error ServerDataBroker::GetFileInfoFromServerById(uint64_t id, FileInfo* info, s
}
std::string ServerDataBroker::GetBeamtimeMeta(Error* err) {
- std::string request_string = "database/" + source_name_ + "/0/meta/0";
- return BrokerRequestWithTimeout(request_string, "", false, err);
+ RequestInfo ri;
+ ri.api = "/database/" + source_name_ + "/0/meta/0";
+
+ return BrokerRequestWithTimeout(ri, err);
+}
+
+FileInfos ServerDataBroker::QueryImages(std::string query, Error* err) {
+ RequestInfo ri;
+ ri.api = "/database/" + source_name_ + "/0/queryimages";
+ ri.post = true;
+ ri.body = std::move(query);
+
+ auto images = BrokerRequestWithTimeout(ri, err);
+
+ return FileInfos{};
}
}
diff --git a/worker/api/cpp/src/server_data_broker.h b/worker/api/cpp/src/server_data_broker.h
index 292477a4b..6662eaae0 100644
--- a/worker/api/cpp/src/server_data_broker.h
+++ b/worker/api/cpp/src/server_data_broker.h
@@ -16,6 +16,15 @@ enum class GetImageServerOperation {
GetID
};
+struct RequestInfo {
+ std::string host;
+ std::string api;
+ std::string extra_params;
+ std::string body;
+ bool post = false;
+};
+
+
class ServerDataBroker final : public asapo::DataBroker {
public:
explicit ServerDataBroker(std::string server_uri, std::string source_path, std::string source_name, std::string token);
@@ -28,6 +37,8 @@ class ServerDataBroker final : public asapo::DataBroker {
uint64_t GetNDataSets(Error* err) override;
Error GetById(uint64_t id, FileInfo* info, std::string group_id, FileData* data) override;
void SetTimeout(uint64_t timeout_ms) override;
+ FileInfos QueryImages(std::string query,Error* err) override;
+
std::unique_ptr<IO> io__; // modified in testings to mock system calls,otherwise do not touch
std::unique_ptr<HttpClient> httpclient__;
std::unique_ptr<NetClient> net_client__;
@@ -38,13 +49,14 @@ class ServerDataBroker final : public asapo::DataBroker {
Error GetDataIfNeeded(FileInfo* info, FileData* data);
Error GetBrokerUri();
void ProcessServerError(Error* err, const std::string& response, std::string* redirect_uri);
- Error ProcessRequest(std::string* response, std::string request_uri, std::string extra_params, bool post);
+ Error ProcessRequest(std::string* response, const RequestInfo& request);
Error GetImageFromServer(GetImageServerOperation op, uint64_t id, std::string group_id, FileInfo* info, FileData* data);
bool DataCanBeInBuffer(const FileInfo* info);
Error TryGetDataFromBuffer(const FileInfo* info, FileData* data);
- std::string BrokerRequestWithTimeout(std::string request_string, std::string extra_params, bool post_request,
- Error* err);
- std::string OpToUriCmd(GetImageServerOperation op);
+ std::string BrokerRequestWithTimeout(RequestInfo request,Error* err);
+ std::string AppendUri(std::string request_string);
+
+ std::string OpToUriCmd(GetImageServerOperation op);
std::string server_uri_;
std::string current_broker_uri_;
std::string source_path_;
diff --git a/worker/api/cpp/unittests/test_folder_broker.cpp b/worker/api/cpp/unittests/test_folder_broker.cpp
index 322089e96..605349beb 100644
--- a/worker/api/cpp/unittests/test_folder_broker.cpp
+++ b/worker/api/cpp/unittests/test_folder_broker.cpp
@@ -344,6 +344,16 @@ TEST_F(GetDataFromFileTests, GetMetaDataReturnsOK) {
ASSERT_THAT(err, Eq(nullptr));
}
+TEST(FolderDataBroker, QueryImages) {
+ auto data_broker = std::unique_ptr<FolderDataBroker>{new FolderDataBroker("test")};
+
+ Error err;
+ auto infos = data_broker->QueryImages("bla",&err);
+
+ ASSERT_THAT(err, Ne(nullptr));
+ ASSERT_THAT(infos.size(), Eq(0));
+}
+
}
diff --git a/worker/api/cpp/unittests/test_server_broker.cpp b/worker/api/cpp/unittests/test_server_broker.cpp
index cd7ba6347..53cb598b2 100644
--- a/worker/api/cpp/unittests/test_server_broker.cpp
+++ b/worker/api/cpp/unittests/test_server_broker.cpp
@@ -63,6 +63,8 @@ class ServerDataBrokerTests : public Test {
std::string expected_full_path = std::string("/tmp/beamline/beamtime") + asapo::kPathSeparator + expected_filename;
std::string expected_group_id = "groupid";
std::string expected_metadata = "{\"meta\":1}";
+ std::string expected_query_string = "bla";
+
uint64_t expected_dataset_id = 1;
static const uint64_t expected_buf_id = 123;
void SetUp() override {
@@ -493,5 +495,25 @@ TEST_F(ServerDataBrokerTests, GetMetaDataOK) {
}
+TEST_F(ServerDataBrokerTests, QueryImagesReturnError) {
+ return;
+ MockGetBrokerUri();
+
+ EXPECT_CALL(mock_http_client, Post_t(HasSubstr("queryimages"), expected_query_string, _, _)).WillOnce(DoAll(
+ SetArgPointee<2>(HttpCode::BadRequest),
+ SetArgPointee<3>(nullptr),
+ Return("error in query")));
+
+ data_broker->SetTimeout(1000);
+ asapo::Error err;
+ auto images = data_broker->QueryImages(expected_query_string,&err);
+
+// ASSERT_THAT(err, Eq(asapo::WorkerErrorTemplates::WrongInput));
+// ASSERT_THAT(err, HasSubstr("query"));
+
+ ASSERT_THAT(images.size(), Eq(0));
+}
+
+
}
--
GitLab