From 7d7ad7d2b61523e6c7dd21ed7169e42d3a53a3a6 Mon Sep 17 00:00:00 2001
From: Sergey Yakubov <sergey.yakubov@desy.de>
Date: Fri, 11 Jun 2021 12:56:28 +0200
Subject: [PATCH] update producer
---
.../cpp/include/asapo/common/data_structs.h | 22 ++++++
common/cpp/include/asapo/common/networking.h | 1 +
.../include/asapo/preprocessor/definitions.h | 11 +++
.../asapo/request/request_handler_factory.h | 1 +
.../unittests/request/test_request_pool.cpp | 1 +
.../api/cpp/src/fabric_consumer_client.cpp | 2 +-
consumer/api/cpp/src/tcp_consumer_client.cpp | 2 +-
.../unittests/test_fabric_consumer_client.cpp | 2 +-
.../unittests/test_tcp_consumer_client.cpp | 2 +-
.../api/cpp/include/asapo/producer/producer.h | 25 +++++-
producer/api/cpp/src/producer_impl.cpp | 46 ++++++++---
producer/api/cpp/src/producer_impl.h | 6 +-
.../src/producer_request_handler_factory.h | 3 +-
.../api/cpp/unittests/test_producer_impl.cpp | 76 ++++++++++++++++---
.../receiver_data_server_request_handler.cpp | 4 +-
receiver/src/request.cpp | 2 +-
.../request_handler/test_request_handler.cpp | 2 +-
receiver/unittests/test_request.cpp | 2 +-
18 files changed, 174 insertions(+), 36 deletions(-)
diff --git a/common/cpp/include/asapo/common/data_structs.h b/common/cpp/include/asapo/common/data_structs.h
index 847650fc7..8fa926288 100644
--- a/common/cpp/include/asapo/common/data_structs.h
+++ b/common/cpp/include/asapo/common/data_structs.h
@@ -145,6 +145,28 @@ enum IngestModeFlags : uint64_t {
const uint64_t kDefaultIngestMode = kTransferData | kStoreInFilesystem | kStoreInDatabase;
+enum class MetaIngestOp: uint64_t {
+ kInsert = 1,
+ kReplace = 2,
+ kUpdate = 3,
+};
+
+struct MetaIngestMode {
+ MetaIngestOp mode;
+ bool upsert;
+ uint64_t Encode(){
+ return static_cast<uint64_t>(mode)+10*static_cast<uint64_t>(upsert);
+ }
+ void Decode(uint64_t code){
+ upsert = code >10;
+ uint64_t val = code-(upsert?10:0);
+ if (val <= 3) {
+ mode=static_cast<MetaIngestOp>(val);
+ }
+ }
+};
+
+
class ClientProtocol {
private:
std::string version_;
diff --git a/common/cpp/include/asapo/common/networking.h b/common/cpp/include/asapo/common/networking.h
index c9e07f1b7..d49d83c32 100644
--- a/common/cpp/include/asapo/common/networking.h
+++ b/common/cpp/include/asapo/common/networking.h
@@ -53,6 +53,7 @@ const std::size_t kNCustomParams = 3;
using CustomRequestData = uint64_t[kNCustomParams];
const std::size_t kPosIngestMode = 0;
const std::size_t kPosDataSetId = 1;
+const std::size_t kPosMetaIngestMode = 1;
const std::size_t kPosDataSetSize = 2;
struct GenericRequestHeader {
diff --git a/common/cpp/include/asapo/preprocessor/definitions.h b/common/cpp/include/asapo/preprocessor/definitions.h
index 3ac042caa..8d78c2b01 100644
--- a/common/cpp/include/asapo/preprocessor/definitions.h
+++ b/common/cpp/include/asapo/preprocessor/definitions.h
@@ -9,6 +9,17 @@
#define FINAL final
#endif
+#if defined(__GNUC__) || defined(__clang__)
+#define DEPRECATED(msg) __attribute__((deprecated(msg)))
+#elif defined(_MSC_VER)
+#define DEPRECATED __declspec(deprecated(msg))
+#else
+#pragma message("WARNING: You need to implement DEPRECATED for this compiler")
+#define DEPRECATED(msg)
+#endif
+
+
+
namespace asapo {
const char kPathSeparator =
#ifdef WIN32
diff --git a/common/cpp/include/asapo/request/request_handler_factory.h b/common/cpp/include/asapo/request/request_handler_factory.h
index c48d787ce..e8cc3e5a6 100644
--- a/common/cpp/include/asapo/request/request_handler_factory.h
+++ b/common/cpp/include/asapo/request/request_handler_factory.h
@@ -10,6 +10,7 @@ namespace asapo {
class RequestHandlerFactory {
public:
virtual std::unique_ptr<RequestHandler> NewRequestHandler(uint64_t thread_id, uint64_t* shared_counter) = 0;
+ virtual ~RequestHandlerFactory(){};
};
diff --git a/common/cpp/unittests/request/test_request_pool.cpp b/common/cpp/unittests/request/test_request_pool.cpp
index f359c1c42..8709d83a1 100644
--- a/common/cpp/unittests/request/test_request_pool.cpp
+++ b/common/cpp/unittests/request/test_request_pool.cpp
@@ -46,6 +46,7 @@ class MockRequestHandlerFactory : public asapo::RequestHandlerFactory {
std::unique_ptr<RequestHandler> NewRequestHandler(uint64_t thread_id, uint64_t* shared_counter) override {
return std::unique_ptr<RequestHandler> {request_handler_};
}
+ ~MockRequestHandlerFactory(){};
private:
RequestHandler* request_handler_;
};
diff --git a/consumer/api/cpp/src/fabric_consumer_client.cpp b/consumer/api/cpp/src/fabric_consumer_client.cpp
index 774513e1f..106104f81 100644
--- a/consumer/api/cpp/src/fabric_consumer_client.cpp
+++ b/consumer/api/cpp/src/fabric_consumer_client.cpp
@@ -34,7 +34,7 @@ Error FabricConsumerClient::GetData(const MessageMeta* info, MessageData* data)
}
GenericRequestHeader request_header{kOpcodeGetBufferData, info->buf_id, info->size};
- strncpy(request_header.api_version, kConsumerProtocol.GetRdsVersion().c_str(), kMaxVersionSize);
+ strncpy(request_header.receiver_protocol, kConsumerProtocol.GetRdsVersion().c_str(), kMaxVersionSize);
memcpy(request_header.message, mr->GetDetails(), sizeof(fabric::MemoryRegionDetails));
GenericNetworkResponse response{};
diff --git a/consumer/api/cpp/src/tcp_consumer_client.cpp b/consumer/api/cpp/src/tcp_consumer_client.cpp
index 7eb9432aa..996ee26fe 100644
--- a/consumer/api/cpp/src/tcp_consumer_client.cpp
+++ b/consumer/api/cpp/src/tcp_consumer_client.cpp
@@ -14,7 +14,7 @@ TcpConsumerClient::TcpConsumerClient() : io__{GenerateDefaultIO()}, connection_p
Error TcpConsumerClient::SendGetDataRequest(SocketDescriptor sd, const MessageMeta* info) const noexcept {
Error err;
GenericRequestHeader request_header{kOpcodeGetBufferData, info->buf_id, info->size};
- strncpy(request_header.api_version, kConsumerProtocol.GetRdsVersion().c_str(), kMaxVersionSize);
+ strncpy(request_header.receiver_protocol, kConsumerProtocol.GetRdsVersion().c_str(), kMaxVersionSize);
io__->Send(sd, &request_header, sizeof(request_header), &err);
if (err) {
connection_pool__->ReleaseConnection(sd);
diff --git a/consumer/api/cpp/unittests/test_fabric_consumer_client.cpp b/consumer/api/cpp/unittests/test_fabric_consumer_client.cpp
index dcbd311e1..23bb6d272 100644
--- a/consumer/api/cpp/unittests/test_fabric_consumer_client.cpp
+++ b/consumer/api/cpp/unittests/test_fabric_consumer_client.cpp
@@ -33,7 +33,7 @@ MATCHER_P6(M_CheckSendRequest, op_code, buf_id, data_size, mr_addr, mr_length, m
&& data->data_id == uint64_t(buf_id)
&& data->data_size == uint64_t(data_size)
&& mr->addr == uint64_t(mr_addr)
- && strcmp(data->api_version, "v0.1") == 0
+ && strcmp(data->receiver_protocol, "v0.1") == 0
&& mr->length == uint64_t(mr_length)
&& mr->key == uint64_t(mr_key);
}
diff --git a/consumer/api/cpp/unittests/test_tcp_consumer_client.cpp b/consumer/api/cpp/unittests/test_tcp_consumer_client.cpp
index ebf4102b3..07789b697 100644
--- a/consumer/api/cpp/unittests/test_tcp_consumer_client.cpp
+++ b/consumer/api/cpp/unittests/test_tcp_consumer_client.cpp
@@ -44,7 +44,7 @@ MATCHER_P4(M_CheckSendRequest, op_code, buf_id, data_size, message,
return ((asapo::GenericRequestHeader*) arg)->op_code == op_code
&& ((asapo::GenericRequestHeader*) arg)->data_id == uint64_t(buf_id)
&& ((asapo::GenericRequestHeader*) arg)->data_size == uint64_t(data_size)
- && strcmp(((asapo::GenericRequestHeader*) arg)->api_version, "v0.1") == 0
+ && strcmp(((asapo::GenericRequestHeader*) arg)->receiver_protocol, "v0.1") == 0
&& strcmp(((asapo::GenericRequestHeader*) arg)->message, message) == 0;
}
diff --git a/producer/api/cpp/include/asapo/producer/producer.h b/producer/api/cpp/include/asapo/producer/producer.h
index 6e35e8b82..53d5ac2e2 100644
--- a/producer/api/cpp/include/asapo/producer/producer.h
+++ b/producer/api/cpp/include/asapo/producer/producer.h
@@ -7,6 +7,7 @@
#include "asapo/logger/logger.h"
#include "common.h"
#include "asapo/common/data_structs.h"
+#include "asapo/preprocessor/definitions.h"
namespace asapo {
@@ -101,16 +102,34 @@ class Producer {
\return Error - Will be nullptr on success
*/
virtual Error SendStreamFinishedFlag(std::string stream, uint64_t last_id, std::string next_stream,
- RequestCallback callback) = 0;
+ RequestCallback callback) = 0 ;
- //! Sends metadata for the current beamtime to the receiver
+ //! Sends beamtime metadata to the receiver
/*!
+ \deprecated { deprecated, obsolates 01.07.2022, use SendBeamtimeMetadata instead}
\param metadata - a JSON string with metadata
\param callback - callback function
\return Error - will be nullptr on success
*/
- virtual Error SendMetadata(const std::string& metadata, RequestCallback callback) = 0;
+ virtual Error SendMetadata(const std::string& metadata, RequestCallback callback) DEPRECATED("obsolates 01.07.2022, use SendBeamtimeMetadata instead") = 0;
+
+ //! Sends beamtime metadata to the receiver
+ /*!
+ \param metadata - a JSON string with metadata
+ \param callback - callback function
+ \return Error - will be nullptr on success
+ */
+ virtual Error SendBeamtimeMetadata(const std::string& metadata, MetaIngestMode mode, RequestCallback callback) = 0;
+
+ //! Sends stream metadata to the receiver
+ /*!
+ \param stream - name of the stream
+ \param metadata - a JSON string with metadata
+ \param callback - callback function
+ \return Error - will be nullptr on success
+ */
+ virtual Error SendStreamMetadata(const std::string& stream, const std::string& metadata, MetaIngestMode mode, RequestCallback callback) = 0;
//! Set internal log level
virtual void SetLogLevel(LogLevel level) = 0;
diff --git a/producer/api/cpp/src/producer_impl.cpp b/producer/api/cpp/src/producer_impl.cpp
index f13f71020..3adc9ae5c 100644
--- a/producer/api/cpp/src/producer_impl.cpp
+++ b/producer/api/cpp/src/producer_impl.cpp
@@ -35,7 +35,7 @@ ProducerImpl::ProducerImpl(std::string endpoint, uint8_t n_processing_threads, u
GenericRequestHeader ProducerImpl::GenerateNextSendRequest(const MessageHeader& message_header, std::string stream,
uint64_t ingest_mode) {
GenericRequestHeader request{kOpcodeTransferData, message_header.message_id, message_header.data_size,
- message_header.user_metadata.size(), message_header.file_name, stream};
+ message_header.user_metadata.size(), message_header.file_name, std::move(stream)};
if (message_header.dataset_substream != 0) {
request.op_code = kOpcodeTransferDatasetData;
request.custom_data[kPosDataSetId] = message_header.dataset_substream;
@@ -251,17 +251,8 @@ Error ProducerImpl::SetCredentials(SourceCredentials source_cred) {
}
Error ProducerImpl::SendMetadata(const std::string& metadata, RequestCallback callback) {
- GenericRequestHeader request_header{kOpcodeTransferMetaData, 0, metadata.size(), 0, "beamtime_global.meta"};
- request_header.custom_data[kPosIngestMode] = asapo::IngestModeFlags::kTransferData |
- asapo::IngestModeFlags::kStoreInDatabase;
- MessageData data{new uint8_t[metadata.size()]};
- strncpy((char*) data.get(), metadata.c_str(), metadata.size());
- auto err = request_pool__->AddRequest(std::unique_ptr<ProducerRequest> {
- new ProducerRequest{
- source_cred_string_, std::move(request_header),
- std::move(data), "", "", callback, true, timeout_ms_}
- });
- return HandleErrorFromPool(std::move(err), true);
+ auto mode=MetaIngestMode{MetaIngestOp::kReplace,true};
+ return SendBeamtimeMetadata(metadata,mode, callback);
}
Error ProducerImpl::Send__(const MessageHeader& message_header,
@@ -483,4 +474,35 @@ Error ProducerImpl::DeleteStream(std::string stream, uint64_t timeout_ms, Delete
return Error{res};
}
+Error ProducerImpl::SendBeamtimeMetadata(const std::string &metadata, MetaIngestMode mode, RequestCallback callback) {
+ return SendMeta("",metadata,mode, callback);
+}
+
+Error ProducerImpl::SendStreamMetadata(const std::string &stream,
+ const std::string &metadata,
+ MetaIngestMode mode,
+ RequestCallback callback) {
+ if (stream.empty()) {
+ return ProducerErrorTemplates::kWrongInput.Generate("stream is empty");
+ }
+ return SendMeta(stream,metadata,mode,callback);
+}
+
+Error ProducerImpl::SendMeta(std::string stream, const std::string &metadata, MetaIngestMode mode,RequestCallback callback) {
+ GenericRequestHeader request_header{kOpcodeTransferMetaData, 0, metadata.size(), 0,
+ stream.empty()?"beamtime_global.meta":stream+".meta",
+ stream};
+ request_header.custom_data[kPosIngestMode] = asapo::IngestModeFlags::kTransferData |
+ asapo::IngestModeFlags::kStoreInDatabase;
+ request_header.custom_data[kPosMetaIngestMode]=mode.Encode();
+ MessageData data{new uint8_t[metadata.size()]};
+ strncpy((char*) data.get(), metadata.c_str(), metadata.size());
+ auto err = request_pool__->AddRequest(std::unique_ptr<ProducerRequest> {
+ new ProducerRequest{
+ source_cred_string_, std::move(request_header),
+ std::move(data), "", "", callback, true, timeout_ms_}
+ });
+ return HandleErrorFromPool(std::move(err), true);
+}
+
}
\ No newline at end of file
diff --git a/producer/api/cpp/src/producer_impl.h b/producer/api/cpp/src/producer_impl.h
index 35d0ff87f..e3de8014b 100644
--- a/producer/api/cpp/src/producer_impl.h
+++ b/producer/api/cpp/src/producer_impl.h
@@ -67,11 +67,15 @@ class ProducerImpl : public Producer {
Error SetCredentials(SourceCredentials source_cred) override;
Error SendMetadata(const std::string& metadata, RequestCallback callback) override;
- uint64_t GetRequestsQueueSize() override;
+ Error SendBeamtimeMetadata(const std::string& metadata, MetaIngestMode mode, RequestCallback callback) override;
+ Error SendStreamMetadata(const std::string& stream, const std::string& metadata, MetaIngestMode mode, RequestCallback callback) override;
+
+ uint64_t GetRequestsQueueSize() override;
Error WaitRequestsFinished(uint64_t timeout_ms) override;
uint64_t GetRequestsQueueVolumeMb() override;
void SetRequestsQueueLimits(uint64_t size, uint64_t volume) override;
private:
+ Error SendMeta(const std::string stream, const std::string &metadata, MetaIngestMode mode, RequestCallback callback);
StreamInfo StreamRequest(StreamRequestOp op, std::string stream, uint64_t timeout_ms, Error* err) const;
Error Send(const MessageHeader& message_header, std::string stream, MessageData data, std::string full_path,
uint64_t ingest_mode,
diff --git a/producer/api/cpp/src/producer_request_handler_factory.h b/producer/api/cpp/src/producer_request_handler_factory.h
index e11f003bf..0200c7c92 100644
--- a/producer/api/cpp/src/producer_request_handler_factory.h
+++ b/producer/api/cpp/src/producer_request_handler_factory.h
@@ -13,7 +13,8 @@ class ProducerRequestHandlerFactory : public RequestHandlerFactory {
public:
ProducerRequestHandlerFactory(ReceiverDiscoveryService* discovery_service);
ProducerRequestHandlerFactory(std::string destination_folder);
- VIRTUAL std::unique_ptr<RequestHandler> NewRequestHandler(uint64_t thread_id, uint64_t* shared_counter) override;
+ virtual std::unique_ptr<RequestHandler> NewRequestHandler(uint64_t thread_id, uint64_t* shared_counter) override;
+ virtual ~ProducerRequestHandlerFactory() { };
private:
RequestHandlerType type_;
ReceiverDiscoveryService* discovery_service_{nullptr};
diff --git a/producer/api/cpp/unittests/test_producer_impl.cpp b/producer/api/cpp/unittests/test_producer_impl.cpp
index 233511dc7..b8cb58dfd 100644
--- a/producer/api/cpp/unittests/test_producer_impl.cpp
+++ b/producer/api/cpp/unittests/test_producer_impl.cpp
@@ -6,6 +6,7 @@
#include "asapo/producer/common.h"
#include "../src/producer_impl.h"
#include "asapo/producer/producer_error.h"
+#include "asapo/common/internal/version.h"
#include "../src/request_handler_tcp.h"
#include "asapo/request/request_pool_error.h"
@@ -39,19 +40,24 @@ MATCHER_P10(M_CheckSendRequest, op_code, source_credentials, metadata, file_id,
dataset_size,
"Checks if a valid GenericRequestHeader was Send") {
auto request = static_cast<ProducerRequest*>(arg);
- return ((asapo::GenericRequestHeader) (arg->header)).op_code == op_code
- && ((asapo::GenericRequestHeader) (arg->header)).data_id == file_id
- && ((asapo::GenericRequestHeader) (arg->header)).data_size == uint64_t(file_size)
+ return ((asapo::GenericRequestHeader) (request->header)).op_code == op_code
+ && ((asapo::GenericRequestHeader) (request->header)).data_id == file_id
+ && ((asapo::GenericRequestHeader) (request->header)).data_size == uint64_t(file_size)
&& request->manage_data_memory == true
&& request->source_credentials == source_credentials
&& request->metadata == metadata
- && (op_code == asapo::kOpcodeTransferDatasetData ? ((asapo::GenericRequestHeader) (arg->header)).custom_data[1]
+ && (op_code == asapo::kOpcodeTransferMetaData ? ((asapo::GenericRequestHeader) (request->header)).custom_data[1]
+ == uint64_t(dataset_id) : true)
+ && (op_code == asapo::kOpcodeTransferDatasetData ? ((asapo::GenericRequestHeader) (request->header)).custom_data[1]
== uint64_t(dataset_id) : true)
- && (op_code == asapo::kOpcodeTransferDatasetData ? ((asapo::GenericRequestHeader) (arg->header)).custom_data[2]
+ && (op_code == asapo::kOpcodeTransferDatasetData ? ((asapo::GenericRequestHeader) (request->header)).custom_data[2]
== uint64_t(dataset_size) : true)
- && ((asapo::GenericRequestHeader) (arg->header)).custom_data[asapo::kPosIngestMode] == uint64_t(ingest_mode)
- && strcmp(((asapo::GenericRequestHeader) (arg->header)).message, message) == 0
- && strcmp(((asapo::GenericRequestHeader) (arg->header)).stream, stream) == 0;
+ && ((asapo::GenericRequestHeader) (request->header)).custom_data[asapo::kPosIngestMode] == uint64_t(ingest_mode)
+ && strcmp(((asapo::GenericRequestHeader) (request->header)).message, message) == 0
+ && strcmp(((asapo::GenericRequestHeader) (request->header)).api_version, asapo::kProducerProtocol.GetReceiverVersion().c_str()) == 0
+ && strcmp(((asapo::GenericRequestHeader) (request->header)).stream, stream) == 0;
+
+
}
TEST(ProducerImpl, Constructor) {
@@ -287,7 +293,7 @@ TEST_F(ProducerImplTests, OKSendingSendDatasetDataRequest) {
ASSERT_THAT(err, Eq(nullptr));
}
-TEST_F(ProducerImplTests, OKAddingSendMetaDataRequest) {
+TEST_F(ProducerImplTests, OKAddingSendMetaDataRequestOld) {
expected_id = 0;
expected_metadata = "{\"meta\":10}";
expected_size = expected_metadata.size();
@@ -302,15 +308,65 @@ TEST_F(ProducerImplTests, OKAddingSendMetaDataRequest) {
"beamtime_global.meta",
"",
expected_ingest_mode,
- 10,
+ 12,
10), false)).WillOnce(Return(
nullptr));
auto err = producer.SendMetadata(expected_metadata, nullptr);
+ ASSERT_THAT(err, Eq(nullptr));
+}
+TEST_F(ProducerImplTests, OKAddingSendMetaDataRequest) {
+ expected_id = 0;
+ expected_metadata = "{\"meta\":10}";
+ expected_size = expected_metadata.size();
+ expected_ingest_mode = asapo::IngestModeFlags::kTransferData | asapo::IngestModeFlags::kStoreInDatabase ;
+
+ producer.SetCredentials(expected_credentials);
+ EXPECT_CALL(mock_pull, AddRequest_t(M_CheckSendRequest(asapo::kOpcodeTransferMetaData,
+ expected_credentials_str,
+ "",
+ expected_id,
+ expected_size,
+ "beamtime_global.meta",
+ "",
+ expected_ingest_mode,
+ 12,
+ 10), false)).WillOnce(Return(
+ nullptr));
+
+ auto mode = asapo::MetaIngestMode{asapo::MetaIngestOp::kReplace,true};
+ auto err = producer.SendBeamtimeMetadata(expected_metadata,mode, nullptr);
ASSERT_THAT(err, Eq(nullptr));
}
+TEST_F(ProducerImplTests, OKAddingSendStreamDataRequest) {
+ expected_id = 0;
+ expected_metadata = "{\"meta\":10}";
+ expected_size = expected_metadata.size();
+ expected_ingest_mode = asapo::IngestModeFlags::kTransferData | asapo::IngestModeFlags::kStoreInDatabase ;
+
+ producer.SetCredentials(expected_credentials);
+ EXPECT_CALL(mock_pull, AddRequest_t(M_CheckSendRequest(asapo::kOpcodeTransferMetaData,
+ expected_credentials_str,
+ "",
+ expected_id,
+ expected_size,
+ (std::string(expected_stream)+".meta").c_str(),
+ expected_stream,
+ expected_ingest_mode,
+ 1,
+ 10), false)).WillOnce(Return(
+ nullptr));
+
+
+ auto mode = asapo::MetaIngestMode{asapo::MetaIngestOp::kInsert,false};
+ auto err = producer.SendStreamMetadata(expected_stream,expected_metadata,mode,nullptr);
+
+ ASSERT_THAT(err, Eq(nullptr));
+}
+
+
TEST_F(ProducerImplTests, ErrorSendingEmptyFileName) {
producer.SetCredentials(expected_credentials);
diff --git a/receiver/src/receiver_data_server/request_handler/receiver_data_server_request_handler.cpp b/receiver/src/receiver_data_server/request_handler/receiver_data_server_request_handler.cpp
index 59406f7f2..00a569b91 100644
--- a/receiver/src/receiver_data_server/request_handler/receiver_data_server_request_handler.cpp
+++ b/receiver/src/receiver_data_server/request_handler/receiver_data_server_request_handler.cpp
@@ -16,7 +16,7 @@ bool ReceiverDataServerRequestHandler::CheckRequest(const ReceiverDataServerRequ
*code = kNetErrorWrongRequest;
return false;
}
- int verClient = VersionToNumber(request->header.api_version);
+ int verClient = VersionToNumber(request->header.receiver_protocol);
int verService = VersionToNumber(GetRdsApiVersion());
if (verClient > verService) {
*code = kNetErrorNotSupported;
@@ -99,7 +99,7 @@ void ReceiverDataServerRequestHandler::HandleInvalidRequest(const ReceiverDataSe
log__->Error("wrong request, code:" + std::to_string(receiver_request->header.op_code));
break;
case NetworkErrorCode::kNetErrorNotSupported:
- log__->Error("unsupported client, version: " + std::string(receiver_request->header.api_version));
+ log__->Error("unsupported client, version: " + std::string(receiver_request->header.receiver_protocol));
break;
};
diff --git a/receiver/src/request.cpp b/receiver/src/request.cpp
index 7db51dfd0..5b35a3d6d 100644
--- a/receiver/src/request.cpp
+++ b/receiver/src/request.cpp
@@ -90,7 +90,7 @@ std::string Request::GetStream() const {
}
std::string Request::GetApiVersion() const {
- return request_header_.api_version;
+ return request_header_.receiver_protocol;
}
diff --git a/receiver/unittests/receiver_data_server/request_handler/test_request_handler.cpp b/receiver/unittests/receiver_data_server/request_handler/test_request_handler.cpp
index fb85f18e5..cceff51bb 100644
--- a/receiver/unittests/receiver_data_server/request_handler/test_request_handler.cpp
+++ b/receiver/unittests/receiver_data_server/request_handler/test_request_handler.cpp
@@ -123,7 +123,7 @@ TEST_F(RequestHandlerTests, ProcessRequest_WrongOpCode) {
}
TEST_F(RequestHandlerTests, ProcessRequest_WrongClientVersion) {
- strcpy(request.header.api_version, "v0.2");
+ strcpy(request.header.receiver_protocol, "v0.2");
MockSendResponse(asapo::kNetErrorNotSupported, false);
EXPECT_CALL(mock_net, HandleAfterError_t(expected_source_id));
diff --git a/receiver/unittests/test_request.cpp b/receiver/unittests/test_request.cpp
index 202884985..5f0a8338c 100644
--- a/receiver/unittests/test_request.cpp
+++ b/receiver/unittests/test_request.cpp
@@ -97,7 +97,7 @@ class RequestTests : public Test {
generic_request_header.op_code = expected_op_code;
generic_request_header.custom_data[asapo::kPosIngestMode] = asapo::kDefaultIngestMode;
strcpy(generic_request_header.message, expected_request_message);
- strcpy(generic_request_header.api_version, expected_api_version.c_str());
+ strcpy(generic_request_header.receiver_protocol, expected_api_version.c_str());
request.reset(new Request{generic_request_header, expected_socket_id, expected_origin_uri, nullptr, nullptr});
request->io__ = std::unique_ptr<asapo::IO> {&mock_io};
ON_CALL(mock_io, Receive_t(expected_socket_id, _, data_size_, _)).WillByDefault(
--
GitLab