data in payload for c++

producer/api/cpp/include/producer/common.h

@@ -14,6 +14,7 @@ const uint8_t kMaxProcessingThreads = 32;
 
 struct RequestCallbackPayload {
     GenericRequestHeader original_header;
+    FileData data;
     std::string response;
 };
 

producer/api/cpp/src/request_handler_filesystem.cpp

@@ -23,6 +23,7 @@ bool RequestHandlerFilesystem::ProcessRequestUnlocked(GenericRequest* request, b
         producer_request->data = io__->GetDataFromFile(producer_request->original_filepath, &producer_request->header.data_size,
                                                        &err);
         if (err) {
+            producer_request->data = nullptr;
             *retry = true;
             return false;
         }
@@ -31,7 +32,7 @@ bool RequestHandlerFilesystem::ProcessRequestUnlocked(GenericRequest* request, b
     err = io__->WriteDataToFile(destination_folder_, request->header.message, (uint8_t*)producer_request->data.get(),
                                 (size_t)request->header.data_size, true, true);
     if (producer_request->callback) {
-        producer_request->callback(RequestCallbackPayload{request->header, ""}, std::move(err));
+        producer_request->callback(RequestCallbackPayload{request->header, std::move(producer_request->data),""}, std::move(err));
     }
     *retry = false;
     return true;

producer/api/cpp/src/request_handler_tcp.cpp

@@ -216,7 +216,7 @@ bool RequestHandlerTcp::ProcessErrorFromReceiver(const Error& error,
 
 void RequestHandlerTcp::ProcessRequestCallback(Error err, ProducerRequest* request, std::string response, bool* retry) {
     if (request->callback) {
-        request->callback(RequestCallbackPayload{request->header, std::move(response)}, std::move(err));
+        request->callback(RequestCallbackPayload{request->header, std::move(request->data),std::move(response)}, std::move(err));
     }
     *retry = false;
 }
@@ -266,7 +266,6 @@ bool RequestHandlerTcp::ProcessRequestUnlocked(GenericRequest* request, bool* re
         return false;
     }
 
-
     if (NeedRebalance()) {
         CloseConnectionToPeformRebalance();
     }
@@ -304,7 +303,7 @@ void RequestHandlerTcp::ProcessRequestTimeout(GenericRequest* request) {
 
     auto err = ProducerErrorTemplates::kTimeout.Generate(err_string);
     if (producer_request->callback) {
-        producer_request->callback(RequestCallbackPayload{request->header, ""}, std::move(err));
+        producer_request->callback(RequestCallbackPayload{request->header, std::move(producer_request->data),""}, std::move(err));
     }
 
 }