You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2022/10/12 15:22:25 UTC
[nifi-minifi-cpp] 02/02: MINIFICPP-1648 Input/OutputStreamCallback should use Input/OutputStream instead of BaseStream
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 7ab10859fbec2a33f9edab4302f0998fe6f123d3
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Oct 12 16:52:21 2022 +0200
MINIFICPP-1648 Input/OutputStreamCallback should use Input/OutputStream instead of BaseStream
+ Change additional basestreams to specific input/output streams
Closes #1428
Signed-off-by: Marton Szasz <sz...@apache.org>
---
controller/MiNiFiController.cpp | 1 -
extensions/aws/processors/FetchS3Object.cpp | 16 ++-----
extensions/aws/processors/PutS3Object.h | 2 +-
extensions/aws/s3/S3Wrapper.cpp | 6 +--
extensions/aws/s3/S3Wrapper.h | 7 ++-
.../azure/processors/FetchAzureBlobStorage.cpp | 2 +-
.../azure/processors/FetchAzureDataLakeStorage.cpp | 2 +-
extensions/azure/processors/PutAzureBlobStorage.h | 2 +-
.../azure/processors/PutAzureDataLakeStorage.cpp | 2 +-
.../azure/processors/PutAzureDataLakeStorage.h | 2 +-
extensions/azure/storage/AzureBlobStorage.cpp | 2 +-
extensions/azure/storage/AzureBlobStorage.h | 2 +-
extensions/azure/storage/AzureDataLakeStorage.cpp | 2 +-
extensions/azure/storage/AzureDataLakeStorage.h | 2 +-
extensions/coap/protocols/CoapC2Protocol.cpp | 6 +--
extensions/coap/protocols/CoapC2Protocol.h | 14 +++---
extensions/coap/tests/CoapC2VerifyHeartbeat.cpp | 1 -
extensions/gcp/processors/FetchGCSObject.cpp | 2 +-
extensions/gcp/processors/PutGCSObject.cpp | 2 +-
extensions/http-curl/client/HTTPCallback.h | 2 +-
extensions/http-curl/sitetosite/HTTPProtocol.cpp | 2 +-
extensions/http-curl/sitetosite/HTTPProtocol.h | 2 +-
extensions/http-curl/tests/CivetStream.h | 1 -
extensions/http-curl/tests/HTTPHandlers.h | 2 +-
.../http-curl/tests/unit/InvokeHTTPTests.cpp | 1 -
extensions/jni/jvm/JniProcessSession.h | 1 -
extensions/jni/jvm/JniReferenceObjects.h | 4 +-
extensions/libarchive/CompressContent.h | 4 +-
extensions/libarchive/FocusArchiveEntry.cpp | 6 +--
extensions/libarchive/FocusArchiveEntry.h | 2 +-
extensions/libarchive/MergeContent.h | 6 +--
extensions/libarchive/UnfocusArchiveEntry.cpp | 4 +-
extensions/libarchive/UnfocusArchiveEntry.h | 2 +-
extensions/librdkafka/PublishKafka.cpp | 4 +-
extensions/mqtt/processors/ConsumeMQTT.cpp | 2 +-
extensions/mqtt/processors/PublishMQTT.cpp | 2 +-
extensions/mqtt/processors/PublishMQTT.h | 2 +-
extensions/opencv/CaptureRTSPFrame.cpp | 14 ++----
extensions/opencv/MotionDetector.cpp | 2 +-
.../lua/{LuaBaseStream.cpp => LuaInputStream.cpp} | 22 +++-------
.../lua/{LuaBaseStream.h => LuaInputStream.h} | 27 +++---------
.../script/lua/LuaOutputStream.cpp | 28 ++++++------
.../lua/{LuaBaseStream.h => LuaOutputStream.h} | 26 +++--------
extensions/script/lua/LuaProcessSession.cpp | 22 +++-------
extensions/script/lua/LuaProcessSession.h | 17 +++-----
extensions/script/lua/LuaScriptEngine.cpp | 10 +++--
.../python/{PyBaseStream.cpp => PyInputStream.cpp} | 12 ++----
.../python/{PyBaseStream.h => PyInputStream.h} | 9 ++--
.../python/{PyBaseStream.h => PyOutputStream.cpp} | 27 +++++-------
.../python/{PyBaseStream.h => PyOutputStream.h} | 10 ++---
extensions/script/python/PyProcessSession.cpp | 8 ++--
extensions/script/python/PyProcessSession.h | 3 +-
extensions/script/python/PythonBindings.h | 13 +++---
extensions/sftp/client/SFTPClient.cpp | 24 +++++------
extensions/sftp/client/SFTPClient.h | 6 +--
extensions/sftp/processors/FetchSFTP.cpp | 7 +--
extensions/sftp/processors/PutSFTP.cpp | 2 +-
.../processors/ExecuteProcess.h | 1 -
.../standard-processors/processors/ExtractText.cpp | 2 +-
.../standard-processors/processors/ExtractText.h | 2 +-
.../standard-processors/processors/HashContent.cpp | 14 ++----
.../standard-processors/processors/HashContent.h | 14 +++---
.../standard-processors/processors/PutFile.cpp | 4 +-
.../standard-processors/processors/PutFile.h | 21 +++------
.../standard-processors/processors/RouteText.cpp | 2 +-
.../standard-processors/processors/TailFile.cpp | 8 ++--
extensions/tensorflow/TFApplyGraph.cpp | 6 +--
extensions/tensorflow/TFApplyGraph.h | 8 ++--
extensions/tensorflow/TFConvertImageToTensor.cpp | 4 +-
extensions/tensorflow/TFConvertImageToTensor.h | 6 ++-
extensions/tensorflow/TFExtractTopLabels.cpp | 4 +-
extensions/tensorflow/TFExtractTopLabels.h | 6 ++-
extensions/usb-camera/GetUSBCamera.cpp | 2 +-
extensions/usb-camera/GetUSBCamera.h | 2 +-
libminifi/include/FlowFileRecord.h | 1 -
libminifi/include/c2/PayloadSerializer.h | 15 ++++---
libminifi/include/core/ContentRepository.h | 1 -
libminifi/include/core/ProcessSession.h | 6 +--
.../include/core/ProcessSessionReadCallback.h | 23 +++-------
libminifi/include/io/BaseStream.h | 12 +-----
libminifi/include/io/CRCStream.h | 3 +-
libminifi/include/io/StreamCallback.h | 9 ++--
libminifi/include/io/StreamPipe.h | 7 +--
libminifi/include/io/StreamSlice.h | 12 +++---
libminifi/include/io/ZlibStream.h | 2 +-
libminifi/include/sitetosite/SiteToSiteClient.h | 19 ++------
libminifi/include/utils/ByteArrayCallback.h | 20 ++-------
libminifi/include/utils/FileReaderCallback.h | 14 ++----
libminifi/include/utils/JsonCallback.h | 6 +--
.../utils/LineByLineInputOutputStreamCallback.h | 5 ++-
libminifi/src/core/ProcessSession.cpp | 14 +++---
libminifi/src/core/ProcessSessionReadCallback.cpp | 21 +++------
libminifi/src/io/StreamSlice.cpp | 2 +-
libminifi/src/sitetosite/SiteToSiteClient.cpp | 6 +--
libminifi/src/utils/ByteArrayCallback.cpp | 29 ++-----------
libminifi/src/utils/FileReaderCallback.cpp | 14 ++----
.../utils/LineByLineInputOutputStreamCallback.cpp | 2 +-
.../test/archive-tests/CompressContentTests.cpp | 2 +-
libminifi/test/archive-tests/MergeFileTests.cpp | 4 +-
.../test/rocksdb-tests/ContentSessionTests.cpp | 4 +-
libminifi/test/rocksdb-tests/SwapTests.cpp | 2 +-
libminifi/test/unit/CRCTests.cpp | 22 +++++-----
.../test/unit/ContentRepositoryDependentTests.h | 2 +-
libminifi/test/unit/FlowFileSerializationTests.cpp | 1 -
libminifi/test/unit/SerializationTests.cpp | 1 -
libminifi/test/unit/StreamTests.cpp | 8 ++--
nanofi/include/cxx/CallbackProcessor.h | 1 -
nanofi/include/cxx/ReflexiveSession.h | 50 +++++++---------------
108 files changed, 325 insertions(+), 534 deletions(-)
diff --git a/controller/MiNiFiController.cpp b/controller/MiNiFiController.cpp
index 6f55e9b45..fd3438859 100644
--- a/controller/MiNiFiController.cpp
+++ b/controller/MiNiFiController.cpp
@@ -23,7 +23,6 @@
#include <queue>
#include <map>
#include <iostream>
-#include "io/BaseStream.h"
#include "core/Core.h"
diff --git a/extensions/aws/processors/FetchS3Object.cpp b/extensions/aws/processors/FetchS3Object.cpp
index 9ee0fd27a..10166f58d 100644
--- a/extensions/aws/processors/FetchS3Object.cpp
+++ b/extensions/aws/processors/FetchS3Object.cpp
@@ -28,12 +28,7 @@
#include "core/Resource.h"
#include "utils/OptionalUtils.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace aws {
-namespace processors {
+namespace org::apache::nifi::minifi::aws::processors {
void FetchS3Object::initialize() {
setSupportedProperties(properties());
@@ -90,7 +85,7 @@ void FetchS3Object::onTrigger(const std::shared_ptr<core::ProcessContext> &conte
}
std::optional<minifi::aws::s3::GetObjectResult> result;
- session->write(flow_file, [&get_object_params, &result, this](const std::shared_ptr<io::BaseStream>& stream) -> int64_t {
+ session->write(flow_file, [&get_object_params, &result, this](const std::shared_ptr<io::OutputStream>& stream) -> int64_t {
result = s3_wrapper_.getObject(*get_object_params, *stream);
return (result | minifi::utils::map(&s3::GetObjectResult::write_size)).value_or(0);
});
@@ -120,9 +115,4 @@ void FetchS3Object::onTrigger(const std::shared_ptr<core::ProcessContext> &conte
}
}
-} // namespace processors
-} // namespace aws
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::aws::processors
diff --git a/extensions/aws/processors/PutS3Object.h b/extensions/aws/processors/PutS3Object.h
index f04f89671..42ef343f3 100644
--- a/extensions/aws/processors/PutS3Object.h
+++ b/extensions/aws/processors/PutS3Object.h
@@ -101,7 +101,7 @@ class PutS3Object : public S3Processor {
, s3_wrapper_(s3_wrapper) {
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
if (flow_size_ > MAX_SIZE) {
return -1;
}
diff --git a/extensions/aws/s3/S3Wrapper.cpp b/extensions/aws/s3/S3Wrapper.cpp
index 6e76676ba..e8f573c8d 100644
--- a/extensions/aws/s3/S3Wrapper.cpp
+++ b/extensions/aws/s3/S3Wrapper.cpp
@@ -74,7 +74,7 @@ std::string S3Wrapper::getEncryptionString(Aws::S3::Model::ServerSideEncryption
return "";
}
-std::optional<PutObjectResult> S3Wrapper::putObject(const PutObjectRequestParameters& put_object_params, std::shared_ptr<Aws::IOStream> data_stream) {
+std::optional<PutObjectResult> S3Wrapper::putObject(const PutObjectRequestParameters& put_object_params, const std::shared_ptr<Aws::IOStream>& data_stream) {
Aws::S3::Model::PutObjectRequest request;
request.SetBucket(put_object_params.bucket);
request.SetKey(put_object_params.object_key);
@@ -116,7 +116,7 @@ bool S3Wrapper::deleteObject(const DeleteObjectRequestParameters& params) {
return request_sender_->sendDeleteObjectRequest(request, params.credentials, params.client_config);
}
-int64_t S3Wrapper::writeFetchedBody(Aws::IOStream& source, const int64_t data_size, io::BaseStream& output) {
+int64_t S3Wrapper::writeFetchedBody(Aws::IOStream& source, const int64_t data_size, io::OutputStream& output) {
std::vector<uint8_t> buffer(4096);
size_t write_size = 0;
if (data_size < 0) return 0;
@@ -134,7 +134,7 @@ int64_t S3Wrapper::writeFetchedBody(Aws::IOStream& source, const int64_t data_si
return gsl::narrow<int64_t>(write_size);
}
-std::optional<GetObjectResult> S3Wrapper::getObject(const GetObjectRequestParameters& get_object_params, io::BaseStream& out_body) {
+std::optional<GetObjectResult> S3Wrapper::getObject(const GetObjectRequestParameters& get_object_params, io::OutputStream& out_body) {
auto request = createFetchObjectRequest<Aws::S3::Model::GetObjectRequest>(get_object_params);
auto aws_result = request_sender_->sendGetObjectRequest(request, get_object_params.credentials, get_object_params.client_config);
if (!aws_result) {
diff --git a/extensions/aws/s3/S3Wrapper.h b/extensions/aws/s3/S3Wrapper.h
index 5424319e1..676857f82 100644
--- a/extensions/aws/s3/S3Wrapper.h
+++ b/extensions/aws/s3/S3Wrapper.h
@@ -40,7 +40,6 @@
#include "utils/StringUtils.h"
#include "utils/ListingStateManager.h"
#include "utils/gsl.h"
-#include "io/BaseStream.h"
#include "S3RequestSender.h"
namespace org::apache::nifi::minifi::aws::s3 {
@@ -200,9 +199,9 @@ class S3Wrapper {
S3Wrapper();
explicit S3Wrapper(std::unique_ptr<S3RequestSender>&& request_sender);
- std::optional<PutObjectResult> putObject(const PutObjectRequestParameters& put_object_params, std::shared_ptr<Aws::IOStream> data_stream);
+ std::optional<PutObjectResult> putObject(const PutObjectRequestParameters& put_object_params, const std::shared_ptr<Aws::IOStream>& data_stream);
bool deleteObject(const DeleteObjectRequestParameters& params);
- std::optional<GetObjectResult> getObject(const GetObjectRequestParameters& get_object_params, io::BaseStream& out_body);
+ std::optional<GetObjectResult> getObject(const GetObjectRequestParameters& get_object_params, io::OutputStream& out_body);
std::optional<std::vector<ListedObjectAttributes>> listBucket(const ListRequestParameters& params);
std::optional<std::map<std::string, std::string>> getObjectTags(const GetObjectTagsParameters& params);
std::optional<HeadObjectResult> headObject(const HeadObjectRequestParameters& head_object_params);
@@ -213,7 +212,7 @@ class S3Wrapper {
static Expiration getExpiration(const std::string& expiration);
void setCannedAcl(Aws::S3::Model::PutObjectRequest& request, const std::string& canned_acl) const;
- static int64_t writeFetchedBody(Aws::IOStream& source, const int64_t data_size, io::BaseStream& output);
+ static int64_t writeFetchedBody(Aws::IOStream& source, const int64_t data_size, io::OutputStream& output);
static std::string getEncryptionString(Aws::S3::Model::ServerSideEncryption encryption);
std::optional<std::vector<ListedObjectAttributes>> listVersions(const ListRequestParameters& params);
diff --git a/extensions/azure/processors/FetchAzureBlobStorage.cpp b/extensions/azure/processors/FetchAzureBlobStorage.cpp
index 0ed7ab951..ecd67e748 100644
--- a/extensions/azure/processors/FetchAzureBlobStorage.cpp
+++ b/extensions/azure/processors/FetchAzureBlobStorage.cpp
@@ -70,7 +70,7 @@ void FetchAzureBlobStorage::onTrigger(const std::shared_ptr<core::ProcessContext
auto fetched_flow_file = session->create(flow_file);
std::optional<int64_t> result_size;
- session->write(fetched_flow_file, [&, this](const std::shared_ptr<io::BaseStream>& stream) -> int64_t {
+ session->write(fetched_flow_file, [&, this](const std::shared_ptr<io::OutputStream>& stream) -> int64_t {
result_size = azure_blob_storage_.fetchBlob(*params, *stream);
if (!result_size) {
return 0;
diff --git a/extensions/azure/processors/FetchAzureDataLakeStorage.cpp b/extensions/azure/processors/FetchAzureDataLakeStorage.cpp
index 06de9ca8d..2b2ff5188 100644
--- a/extensions/azure/processors/FetchAzureDataLakeStorage.cpp
+++ b/extensions/azure/processors/FetchAzureDataLakeStorage.cpp
@@ -74,7 +74,7 @@ void FetchAzureDataLakeStorage::onTrigger(const std::shared_ptr<core::ProcessCon
auto fetched_flow_file = session->create(flow_file);
std::optional<uint64_t> result;
- session->write(fetched_flow_file, [&, this](const std::shared_ptr<io::BaseStream>& output_stream) -> int64_t {
+ session->write(fetched_flow_file, [&, this](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
result = azure_data_lake_storage_.fetchFile(*params, *output_stream);
if (!result) {
return 0;
diff --git a/extensions/azure/processors/PutAzureBlobStorage.h b/extensions/azure/processors/PutAzureBlobStorage.h
index 619269f04..3f9574deb 100644
--- a/extensions/azure/processors/PutAzureBlobStorage.h
+++ b/extensions/azure/processors/PutAzureBlobStorage.h
@@ -73,7 +73,7 @@ class PutAzureBlobStorage final : public AzureBlobStorageSingleBlobProcessorBase
, params_(params) {
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
std::vector<std::byte> buffer;
buffer.resize(flow_size_);
size_t read_ret = stream->read(buffer);
diff --git a/extensions/azure/processors/PutAzureDataLakeStorage.cpp b/extensions/azure/processors/PutAzureDataLakeStorage.cpp
index ccf5c6754..027d51524 100644
--- a/extensions/azure/processors/PutAzureDataLakeStorage.cpp
+++ b/extensions/azure/processors/PutAzureDataLakeStorage.cpp
@@ -116,7 +116,7 @@ PutAzureDataLakeStorage::ReadCallback::ReadCallback(
logger_(std::move(logger)) {
}
-int64_t PutAzureDataLakeStorage::ReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t PutAzureDataLakeStorage::ReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) {
std::vector<std::byte> buffer;
buffer.resize(flow_size_);
size_t read_ret = stream->read(buffer);
diff --git a/extensions/azure/processors/PutAzureDataLakeStorage.h b/extensions/azure/processors/PutAzureDataLakeStorage.h
index ac674dbcf..8fa3fffc4 100644
--- a/extensions/azure/processors/PutAzureDataLakeStorage.h
+++ b/extensions/azure/processors/PutAzureDataLakeStorage.h
@@ -75,7 +75,7 @@ class PutAzureDataLakeStorage final : public AzureDataLakeStorageFileProcessorBa
class ReadCallback {
public:
ReadCallback(uint64_t flow_size, storage::AzureDataLakeStorage& azure_data_lake_storage, const storage::PutAzureDataLakeStorageParameters& params, std::shared_ptr<core::logging::Logger> logger);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream);
[[nodiscard]] storage::UploadDataLakeStorageResult getResult() const {
return result_;
diff --git a/extensions/azure/storage/AzureBlobStorage.cpp b/extensions/azure/storage/AzureBlobStorage.cpp
index d881ecdb1..a5d9ec5b7 100644
--- a/extensions/azure/storage/AzureBlobStorage.cpp
+++ b/extensions/azure/storage/AzureBlobStorage.cpp
@@ -75,7 +75,7 @@ bool AzureBlobStorage::deleteBlob(const DeleteAzureBlobStorageParameters& params
}
}
-std::optional<uint64_t> AzureBlobStorage::fetchBlob(const FetchAzureBlobStorageParameters& params, io::BaseStream& stream) {
+std::optional<uint64_t> AzureBlobStorage::fetchBlob(const FetchAzureBlobStorageParameters& params, io::OutputStream& stream) {
try {
auto fetch_res = blob_storage_client_->fetchBlob(params);
return internal::pipe(*fetch_res, stream);
diff --git a/extensions/azure/storage/AzureBlobStorage.h b/extensions/azure/storage/AzureBlobStorage.h
index 5b1acb804..8b7f9b16e 100644
--- a/extensions/azure/storage/AzureBlobStorage.h
+++ b/extensions/azure/storage/AzureBlobStorage.h
@@ -66,7 +66,7 @@ class AzureBlobStorage {
std::optional<bool> createContainerIfNotExists(const PutAzureBlobStorageParameters& params);
std::optional<UploadBlobResult> uploadBlob(const PutAzureBlobStorageParameters& params, gsl::span<const std::byte> buffer);
bool deleteBlob(const DeleteAzureBlobStorageParameters& params);
- std::optional<uint64_t> fetchBlob(const FetchAzureBlobStorageParameters& params, io::BaseStream& stream);
+ std::optional<uint64_t> fetchBlob(const FetchAzureBlobStorageParameters& params, io::OutputStream& stream);
std::optional<ListContainerResult> listContainer(const ListAzureBlobStorageParameters& params);
private:
diff --git a/extensions/azure/storage/AzureDataLakeStorage.cpp b/extensions/azure/storage/AzureDataLakeStorage.cpp
index 1546b8f05..b53bc201a 100644
--- a/extensions/azure/storage/AzureDataLakeStorage.cpp
+++ b/extensions/azure/storage/AzureDataLakeStorage.cpp
@@ -92,7 +92,7 @@ bool AzureDataLakeStorage::deleteFile(const DeleteAzureDataLakeStorageParameters
}
}
-std::optional<uint64_t> AzureDataLakeStorage::fetchFile(const FetchAzureDataLakeStorageParameters& params, io::BaseStream& stream) {
+std::optional<uint64_t> AzureDataLakeStorage::fetchFile(const FetchAzureDataLakeStorageParameters& params, io::OutputStream& stream) {
try {
auto result = data_lake_storage_client_->fetchFile(params);
return internal::pipe(*result, stream);
diff --git a/extensions/azure/storage/AzureDataLakeStorage.h b/extensions/azure/storage/AzureDataLakeStorage.h
index 8495a4d18..d7e05f7db 100644
--- a/extensions/azure/storage/AzureDataLakeStorage.h
+++ b/extensions/azure/storage/AzureDataLakeStorage.h
@@ -69,7 +69,7 @@ class AzureDataLakeStorage {
storage::UploadDataLakeStorageResult uploadFile(const storage::PutAzureDataLakeStorageParameters& params, gsl::span<const std::byte> buffer);
bool deleteFile(const storage::DeleteAzureDataLakeStorageParameters& params);
- std::optional<uint64_t> fetchFile(const FetchAzureDataLakeStorageParameters& params, io::BaseStream& stream);
+ std::optional<uint64_t> fetchFile(const FetchAzureDataLakeStorageParameters& params, io::OutputStream& stream);
std::optional<ListDataLakeStorageResult> listDirectory(const ListAzureDataLakeStorageParameters& params);
private:
diff --git a/extensions/coap/protocols/CoapC2Protocol.cpp b/extensions/coap/protocols/CoapC2Protocol.cpp
index 74c4421f5..a08b5fd72 100644
--- a/extensions/coap/protocols/CoapC2Protocol.cpp
+++ b/extensions/coap/protocols/CoapC2Protocol.cpp
@@ -19,7 +19,7 @@
#include "c2/PayloadParser.h"
#include "coap_functions.h"
#include "coap_message.h"
-#include "io/BaseStream.h"
+#include "io/OutputStream.h"
#include "core/Resource.h"
#include "utils/gsl.h"
@@ -50,7 +50,7 @@ minifi::c2::C2Payload CoapProtocol::consumePayload(const std::string &url, const
return RESTSender::consumePayload(url, payload, direction, false);
}
-int CoapProtocol::writeAcknowledgement(io::BaseStream *stream, const minifi::c2::C2Payload &payload) {
+int CoapProtocol::writeAcknowledgement(io::OutputStream *stream, const minifi::c2::C2Payload &payload) {
auto ident = payload.getIdentifier();
auto state = payload.getStatus().getState();
stream->write(ident);
@@ -78,7 +78,7 @@ int CoapProtocol::writeAcknowledgement(io::BaseStream *stream, const minifi::c2:
return 0;
}
-int CoapProtocol::writeHeartbeat(io::BaseStream *stream, const minifi::c2::C2Payload &payload) {
+int CoapProtocol::writeHeartbeat(io::OutputStream *stream, const minifi::c2::C2Payload &payload) {
bool byte;
uint16_t size = 0;
diff --git a/extensions/coap/protocols/CoapC2Protocol.h b/extensions/coap/protocols/CoapC2Protocol.h
index c39275522..b1b9de93d 100644
--- a/extensions/coap/protocols/CoapC2Protocol.h
+++ b/extensions/coap/protocols/CoapC2Protocol.h
@@ -28,7 +28,7 @@
#include <mutex>
#include "c2/C2Protocol.h"
-#include "io/BaseStream.h"
+#include "io/OutputStream.h"
#include "agent/agent_version.h"
#include "CoapConnector.h"
@@ -92,20 +92,20 @@ class CoapProtocol : public minifi::c2::RESTSender {
static minifi::c2::Operation getOperation(int type);
/**
- * Writes a heartbeat to the provided BaseStream ptr.
- * @param stream BaseStream
+ * Writes a heartbeat to the provided OutputStream ptr.
+ * @param stream OutputStream
* @param payload payload to serialize
* @return result 0 if success failure otherwise
*/
- int writeHeartbeat(io::BaseStream *stream, const minifi::c2::C2Payload &payload);
+ int writeHeartbeat(io::OutputStream *stream, const minifi::c2::C2Payload &payload);
/**
- * Writes a acknowledgement to the provided BaseStream ptr.
- * @param stream BaseStream
+ * Writes a acknowledgement to the provided OutputStream ptr.
+ * @param stream OutputStream
* @param payload payload to serialize
* @return result 0 if success failure otherwise
*/
- static int writeAcknowledgement(io::BaseStream *stream, const minifi::c2::C2Payload &payload);
+ static int writeAcknowledgement(io::OutputStream *stream, const minifi::c2::C2Payload &payload);
minifi::c2::C2Payload serialize(const minifi::c2::C2Payload &payload);
diff --git a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp b/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
index 6bc87aaec..703f5cff3 100644
--- a/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
+++ b/extensions/coap/tests/CoapC2VerifyHeartbeat.cpp
@@ -52,7 +52,6 @@
#include "processors/LogAttribute.h"
#include "CoapC2Protocol.h"
#include "CoapServer.h"
-#include "io/BaseStream.h"
#include "concurrentqueue.h"
#include "utils/IntegrationTestUtils.h"
diff --git a/extensions/gcp/processors/FetchGCSObject.cpp b/extensions/gcp/processors/FetchGCSObject.cpp
index 33f4b85d9..13ad382a3 100644
--- a/extensions/gcp/processors/FetchGCSObject.cpp
+++ b/extensions/gcp/processors/FetchGCSObject.cpp
@@ -37,7 +37,7 @@ class FetchFromGCSCallback {
client_(client) {
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) {
auto reader = client_.ReadObject(bucket_, key_, encryption_key_, generation_, gcs::IfGenerationNotMatch(0));
auto set_members = gsl::finally([&]{
status_ = reader.status();
diff --git a/extensions/gcp/processors/PutGCSObject.cpp b/extensions/gcp/processors/PutGCSObject.cpp
index 2bee2ae43..52a3c433f 100644
--- a/extensions/gcp/processors/PutGCSObject.cpp
+++ b/extensions/gcp/processors/PutGCSObject.cpp
@@ -37,7 +37,7 @@ class UploadToGCSCallback {
client_(client) {
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
std::string content;
content.resize(stream->size());
const auto read_ret = stream->read(gsl::make_span(content).as_span<std::byte>());
diff --git a/extensions/http-curl/client/HTTPCallback.h b/extensions/http-curl/client/HTTPCallback.h
index 9c4a20c08..8f95712bc 100644
--- a/extensions/http-curl/client/HTTPCallback.h
+++ b/extensions/http-curl/client/HTTPCallback.h
@@ -65,7 +65,7 @@ class HttpStreamingCallback final : public utils::HTTPUploadCallback {
seekInner(lock, pos);
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) override {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) override {
std::vector<std::byte> vec;
if (stream->size() > 0) {
diff --git a/extensions/http-curl/sitetosite/HTTPProtocol.cpp b/extensions/http-curl/sitetosite/HTTPProtocol.cpp
index 9c7b01809..4e713bd45 100644
--- a/extensions/http-curl/sitetosite/HTTPProtocol.cpp
+++ b/extensions/http-curl/sitetosite/HTTPProtocol.cpp
@@ -164,7 +164,7 @@ int HttpSiteToSiteClient::readResponse(const std::shared_ptr<sitetosite::Transac
return SiteToSiteClient::readResponse(transaction, code, message);
}
// write respond
-int HttpSiteToSiteClient::writeResponse(const std::shared_ptr<sitetosite::Transaction> &transaction, sitetosite::RespondCode code, std::string message) {
+int HttpSiteToSiteClient::writeResponse(const std::shared_ptr<sitetosite::Transaction> &transaction, sitetosite::RespondCode code, const std::string& message) {
current_code = code;
if (code == sitetosite::CONFIRM_TRANSACTION || code == sitetosite::FINISH_TRANSACTION) {
return 1;
diff --git a/extensions/http-curl/sitetosite/HTTPProtocol.h b/extensions/http-curl/sitetosite/HTTPProtocol.h
index 19fd8b38a..b313f7993 100644
--- a/extensions/http-curl/sitetosite/HTTPProtocol.h
+++ b/extensions/http-curl/sitetosite/HTTPProtocol.h
@@ -88,7 +88,7 @@ class HttpSiteToSiteClient : public sitetosite::SiteToSiteClient {
int readResponse(const std::shared_ptr<sitetosite::Transaction> &transaction, sitetosite::RespondCode &code, std::string &message) override;
// write respond
- int writeResponse(const std::shared_ptr<sitetosite::Transaction> &transaction, sitetosite::RespondCode code, std::string message) override;
+ int writeResponse(const std::shared_ptr<sitetosite::Transaction> &transaction, sitetosite::RespondCode code, const std::string& message) override;
/**
* Bootstrapping is not really required for the HTTP Site To Site so we will set the peer state and return true.
diff --git a/extensions/http-curl/tests/CivetStream.h b/extensions/http-curl/tests/CivetStream.h
index 16d5c3c5d..a963503ee 100644
--- a/extensions/http-curl/tests/CivetStream.h
+++ b/extensions/http-curl/tests/CivetStream.h
@@ -19,7 +19,6 @@
#include <memory>
-#include "io/BaseStream.h"
#include "civetweb.h"
#include "utils/gsl.h"
diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h
index a49090e72..c0f8a9ad2 100644
--- a/extensions/http-curl/tests/HTTPHandlers.h
+++ b/extensions/http-curl/tests/HTTPHandlers.h
@@ -301,7 +301,7 @@ class FlowFileResponder : public ServerAwareHandler {
"Connection: close\r\n\r\n",
total);
minifi::io::BufferStream serializer;
- minifi::io::CRCStream < minifi::io::BaseStream > stream(gsl::make_not_null(&serializer));
+ minifi::io::CRCStream <minifi::io::OutputStream> stream(gsl::make_not_null(&serializer));
for (const auto& flow : flows) {
uint32_t num_attributes = gsl::narrow<uint32_t>(flow->attributes.size());
stream.write(num_attributes);
diff --git a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
index 9d04be6c4..4fcad4071 100644
--- a/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
+++ b/extensions/http-curl/tests/unit/InvokeHTTPTests.cpp
@@ -19,7 +19,6 @@
#include <memory>
#include <utility>
#include <string>
-#include "io/BaseStream.h"
#include "TestBase.h"
#include "Catch.h"
#include "core/Core.h"
diff --git a/extensions/jni/jvm/JniProcessSession.h b/extensions/jni/jvm/JniProcessSession.h
index 8bd286427..534466b3f 100644
--- a/extensions/jni/jvm/JniProcessSession.h
+++ b/extensions/jni/jvm/JniProcessSession.h
@@ -25,7 +25,6 @@
#include <iterator>
#include <algorithm>
-#include "io/BaseStream.h"
#include "FlowFileRecord.h"
#include "core/ProcessSession.h"
#include "core/ProcessSessionFactory.h"
diff --git a/extensions/jni/jvm/JniReferenceObjects.h b/extensions/jni/jvm/JniReferenceObjects.h
index 366e89546..1d16a1f81 100644
--- a/extensions/jni/jvm/JniReferenceObjects.h
+++ b/extensions/jni/jvm/JniReferenceObjects.h
@@ -105,7 +105,7 @@ class JniByteInputStream {
: buffer_(size),
read_size_(0) {
}
- int64_t operator()(const std::shared_ptr<minifi::io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<minifi::io::InputStream>& stream) {
stream_ = stream;
return 0;
}
@@ -145,7 +145,7 @@ class JniByteInputStream {
return stream_->read(arr);
}
- std::shared_ptr<minifi::io::BaseStream> stream_;
+ std::shared_ptr<minifi::io::InputStream> stream_;
std::vector<std::byte> buffer_;
uint64_t read_size_;
};
diff --git a/extensions/libarchive/CompressContent.h b/extensions/libarchive/CompressContent.h
index 9694fe30d..d3ac8531e 100644
--- a/extensions/libarchive/CompressContent.h
+++ b/extensions/libarchive/CompressContent.h
@@ -112,14 +112,14 @@ class CompressContent : public core::Processor {
std::shared_ptr<core::ProcessSession> session_;
bool success_{false};
- int64_t operator()(const std::shared_ptr<io::BaseStream>& output_stream) {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& output_stream) {
std::shared_ptr<io::ZlibBaseStream> filterStream;
if (compress_mode_ == CompressionMode::Compress) {
filterStream = std::make_shared<io::ZlibCompressStream>(gsl::make_not_null(output_stream.get()), io::ZlibCompressionFormat::GZIP, compress_level_);
} else {
filterStream = std::make_shared<io::ZlibDecompressStream>(gsl::make_not_null(output_stream.get()), io::ZlibCompressionFormat::GZIP);
}
- session_->read(flow_, [this, &filterStream](const std::shared_ptr<io::BaseStream>& input_stream) -> int64_t {
+ session_->read(flow_, [this, &filterStream](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
std::vector<std::byte> buffer(16 * 1024U);
size_t read_size = 0;
while (read_size < flow_->getSize()) {
diff --git a/extensions/libarchive/FocusArchiveEntry.cpp b/extensions/libarchive/FocusArchiveEntry.cpp
index 9d11dd7c9..cd057f2d9 100644
--- a/extensions/libarchive/FocusArchiveEntry.cpp
+++ b/extensions/libarchive/FocusArchiveEntry.cpp
@@ -85,7 +85,7 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSe
}
// Restore target archive entry
- if (targetEntryStashKey != "") {
+ if (!targetEntryStashKey.empty()) {
session->restore(targetEntryStashKey, flowFile);
} else {
logger_->log_warn("FocusArchiveEntry failed to locate target entry: %s",
@@ -127,7 +127,7 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSe
}
struct FocusArchiveEntryReadData {
- std::shared_ptr<io::BaseStream> stream;
+ std::shared_ptr<io::InputStream> stream;
core::Processor *processor;
std::array<std::byte, 8196> buf;
};
@@ -152,7 +152,7 @@ la_ssize_t FocusArchiveEntry::ReadCallback::read_cb(struct archive * a, void *d,
return gsl::narrow<la_ssize_t>(read);
}
-int64_t FocusArchiveEntry::ReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+int64_t FocusArchiveEntry::ReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) const {
auto inputArchive = archive_read_new();
struct archive_entry *entry;
int64_t nlen = 0;
diff --git a/extensions/libarchive/FocusArchiveEntry.h b/extensions/libarchive/FocusArchiveEntry.h
index 8db05eeb4..944de6661 100644
--- a/extensions/libarchive/FocusArchiveEntry.h
+++ b/extensions/libarchive/FocusArchiveEntry.h
@@ -62,7 +62,7 @@ class FocusArchiveEntry : public core::Processor {
class ReadCallback {
public:
explicit ReadCallback(core::Processor*, utils::file::FileManager *file_man, ArchiveMetadata *archiveMetadata);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const;
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) const;
bool isRunning() {return proc_->isRunning();}
private:
diff --git a/extensions/libarchive/MergeContent.h b/extensions/libarchive/MergeContent.h
index 40bf1096b..7880c13da 100644
--- a/extensions/libarchive/MergeContent.h
+++ b/extensions/libarchive/MergeContent.h
@@ -78,7 +78,7 @@ class BinaryConcatenationMerge : public MergeBin {
std::deque<std::shared_ptr<core::FlowFile>> &flows_;
FlowFileSerializer& serializer_;
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) const {
size_t write_size_sum = 0;
if (!header_.empty()) {
const auto write_ret = stream->write(reinterpret_cast<const uint8_t*>(header_.data()), header_.size());
@@ -164,7 +164,7 @@ class ArchiveMerge {
std::string merge_type_;
std::deque<std::shared_ptr<core::FlowFile>> &flows_;
- std::shared_ptr<io::BaseStream> stream_;
+ std::shared_ptr<io::OutputStream> stream_;
size_t size_;
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ArchiveMerge>::getLogger();
FlowFileSerializer& serializer_;
@@ -190,7 +190,7 @@ class ArchiveMerge {
return totalWrote;
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) {
struct archive *arch;
arch = archive_write_new();
diff --git a/extensions/libarchive/UnfocusArchiveEntry.cpp b/extensions/libarchive/UnfocusArchiveEntry.cpp
index 02222b98b..fe1f86f36 100644
--- a/extensions/libarchive/UnfocusArchiveEntry.cpp
+++ b/extensions/libarchive/UnfocusArchiveEntry.cpp
@@ -140,7 +140,7 @@ UnfocusArchiveEntry::WriteCallback::WriteCallback(ArchiveMetadata *archiveMetada
}
struct UnfocusArchiveEntryWriteData {
- std::shared_ptr<io::BaseStream> stream;
+ std::shared_ptr<io::OutputStream> stream;
};
la_ssize_t UnfocusArchiveEntry::WriteCallback::write_cb(struct archive *, void *d, const void *buffer, size_t length) {
@@ -149,7 +149,7 @@ la_ssize_t UnfocusArchiveEntry::WriteCallback::write_cb(struct archive *, void *
return io::isError(write_ret) ? -1 : gsl::narrow<la_ssize_t>(write_ret);
}
-int64_t UnfocusArchiveEntry::WriteCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+int64_t UnfocusArchiveEntry::WriteCallback::operator()(const std::shared_ptr<io::OutputStream>& stream) const {
auto outputArchive = archive_write_new();
int64_t nlen = 0;
diff --git a/extensions/libarchive/UnfocusArchiveEntry.h b/extensions/libarchive/UnfocusArchiveEntry.h
index 587109afa..65733817a 100644
--- a/extensions/libarchive/UnfocusArchiveEntry.h
+++ b/extensions/libarchive/UnfocusArchiveEntry.h
@@ -64,7 +64,7 @@ class UnfocusArchiveEntry : public core::Processor {
class WriteCallback {
public:
explicit WriteCallback(ArchiveMetadata *archiveMetadata);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const;
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) const;
private:
//! Logger
std::shared_ptr<Logger> logger_ = core::logging::LoggerFactory<UnfocusArchiveEntry>::getLogger();
diff --git a/extensions/librdkafka/PublishKafka.cpp b/extensions/librdkafka/PublishKafka.cpp
index 470ed661c..6c141fc24 100644
--- a/extensions/librdkafka/PublishKafka.cpp
+++ b/extensions/librdkafka/PublishKafka.cpp
@@ -258,7 +258,7 @@ class ReadCallback {
ReadCallback(const ReadCallback&) = delete;
ReadCallback& operator=(ReadCallback) = delete;
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
std::vector<std::byte> buffer;
buffer.resize(max_seg_size_);
@@ -734,7 +734,7 @@ void PublishKafka::onTrigger(const std::shared_ptr<core::ProcessContext> &contex
context->getProperty(FailEmptyFlowFiles.getName(), failEmptyFlowFiles);
ReadCallback callback(max_flow_seg_size_, kafkaKey, thisTopic->getTopic(), conn_->getConnection(), *flowFile,
- attributeNameRegex_, messages, flow_file_index, failEmptyFlowFiles, logger_);
+ attributeNameRegex_, messages, flow_file_index, failEmptyFlowFiles, logger_);
session->read(flowFile, std::ref(callback));
if (!callback.called_) {
diff --git a/extensions/mqtt/processors/ConsumeMQTT.cpp b/extensions/mqtt/processors/ConsumeMQTT.cpp
index 2de3fdf08..ba82871ff 100644
--- a/extensions/mqtt/processors/ConsumeMQTT.cpp
+++ b/extensions/mqtt/processors/ConsumeMQTT.cpp
@@ -80,7 +80,7 @@ void ConsumeMQTT::onTrigger(const std::shared_ptr<core::ProcessContext>& /*conte
const auto& message = msg_queue.front();
std::shared_ptr<core::FlowFile> processFlowFile = session->create();
int write_status{};
- session->write(processFlowFile, [&message, &write_status](const std::shared_ptr<io::BaseStream>& stream) -> int64_t {
+ session->write(processFlowFile, [&message, &write_status](const std::shared_ptr<io::OutputStream>& stream) -> int64_t {
if (message->payloadlen < 0) {
write_status = -1;
return -1;
diff --git a/extensions/mqtt/processors/PublishMQTT.cpp b/extensions/mqtt/processors/PublishMQTT.cpp
index e73544016..218a4220a 100644
--- a/extensions/mqtt/processors/PublishMQTT.cpp
+++ b/extensions/mqtt/processors/PublishMQTT.cpp
@@ -70,7 +70,7 @@ void PublishMQTT::onTrigger(const std::shared_ptr<core::ProcessContext>& /*conte
}
}
-int64_t PublishMQTT::ReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t PublishMQTT::ReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) {
if (flow_size_ < max_seg_size_)
max_seg_size_ = flow_size_;
gsl_Expects(max_seg_size_ < gsl::narrow<uint64_t>(std::numeric_limits<int>::max()));
diff --git a/extensions/mqtt/processors/PublishMQTT.h b/extensions/mqtt/processors/PublishMQTT.h
index f940b73b5..38736ba2d 100644
--- a/extensions/mqtt/processors/PublishMQTT.h
+++ b/extensions/mqtt/processors/PublishMQTT.h
@@ -74,7 +74,7 @@ class PublishMQTT : public processors::AbstractMQTTProcessor {
retain_(retain) {
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream);
size_t read_size_ = 0;
int status_ = 0;
diff --git a/extensions/opencv/CaptureRTSPFrame.cpp b/extensions/opencv/CaptureRTSPFrame.cpp
index 8f1d87a1c..eb0968ade 100644
--- a/extensions/opencv/CaptureRTSPFrame.cpp
+++ b/extensions/opencv/CaptureRTSPFrame.cpp
@@ -20,11 +20,7 @@
#include "core/ProcessSession.h"
#include "core/Resource.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace processors {
+namespace org::apache::nifi::minifi::processors {
const core::Property CaptureRTSPFrame::RTSPUsername(
"RTSP Username",
@@ -140,7 +136,7 @@ void CaptureRTSPFrame::onTrigger(const std::shared_ptr<core::ProcessContext> &co
session->putAttribute(flow_file, "filename", filename);
session->putAttribute(flow_file, "video.backend.driver", video_backend_driver_);
- session->write(flow_file, [&frame, this](const std::shared_ptr<io::BaseStream>& output_stream) -> int64_t {
+ session->write(flow_file, [&frame, this](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
std::vector<uchar> image_buf;
imencode(image_encoding_, frame, image_buf);
const auto ret = output_stream->write(image_buf.data(), image_buf.size());
@@ -163,8 +159,4 @@ void CaptureRTSPFrame::notifyStop() {
REGISTER_RESOURCE(CaptureRTSPFrame, Processor);
-} /* namespace processors */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::processors
diff --git a/extensions/opencv/MotionDetector.cpp b/extensions/opencv/MotionDetector.cpp
index 2a1d5bb17..75c7cf5be 100644
--- a/extensions/opencv/MotionDetector.cpp
+++ b/extensions/opencv/MotionDetector.cpp
@@ -152,7 +152,7 @@ void MotionDetector::onTrigger(const std::shared_ptr<core::ProcessContext> &cont
}
cv::Mat frame;
- session->read(flow_file, [&frame](const std::shared_ptr<io::BaseStream>& input_stream) -> int64_t {
+ session->read(flow_file, [&frame](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
std::vector<uchar> image_buf;
image_buf.resize(input_stream->size());
const auto ret = input_stream->read(gsl::make_span(image_buf).as_span<std::byte>());
diff --git a/extensions/script/lua/LuaBaseStream.cpp b/extensions/script/lua/LuaInputStream.cpp
similarity index 77%
rename from extensions/script/lua/LuaBaseStream.cpp
rename to extensions/script/lua/LuaInputStream.cpp
index 76b75619e..5e7e4a466 100644
--- a/extensions/script/lua/LuaBaseStream.cpp
+++ b/extensions/script/lua/LuaInputStream.cpp
@@ -21,19 +21,15 @@
#include <utility>
#include <string>
-#include "LuaBaseStream.h"
+#include "LuaInputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace lua {
+namespace org::apache::nifi::minifi::lua {
-LuaBaseStream::LuaBaseStream(std::shared_ptr<io::BaseStream> stream)
+LuaInputStream::LuaInputStream(std::shared_ptr<io::InputStream> stream)
: stream_(std::move(stream)) {
}
-std::string LuaBaseStream::read(size_t len) {
+std::string LuaInputStream::read(size_t len) {
if (len == 0) {
len = stream_->size();
}
@@ -59,12 +55,4 @@ std::string LuaBaseStream::read(size_t len) {
return io::isError(read) ? std::string{} : buffer;
}
-size_t LuaBaseStream::write(std::string buf) {
- return stream_->write(reinterpret_cast<const uint8_t*>(buf.data()), buf.length());
-}
-
-} /* namespace lua */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/extensions/script/lua/LuaBaseStream.h b/extensions/script/lua/LuaInputStream.h
similarity index 67%
copy from extensions/script/lua/LuaBaseStream.h
copy to extensions/script/lua/LuaInputStream.h
index 47b5017df..24df118c1 100644
--- a/extensions/script/lua/LuaBaseStream.h
+++ b/extensions/script/lua/LuaInputStream.h
@@ -21,17 +21,13 @@
#include <string>
#include "sol/sol.hpp"
-#include "io/BaseStream.h"
+#include "io/InputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace lua {
+namespace org::apache::nifi::minifi::lua {
-class LuaBaseStream {
+class LuaInputStream {
public:
- explicit LuaBaseStream(std::shared_ptr<io::BaseStream> stream);
+ explicit LuaInputStream(std::shared_ptr<io::InputStream> stream);
/**
* Read n bytes of data (returns string, to follow Lua idioms)
@@ -39,19 +35,8 @@ class LuaBaseStream {
*/
std::string read(size_t len = 0);
- /**
- * Write data (receives string, to follow Lua idioms)
- * @param buf
- * @return
- */
- size_t write(std::string buf);
-
private:
- std::shared_ptr<io::BaseStream> stream_;
+ std::shared_ptr<io::InputStream> stream_;
};
-} /* namespace lua */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/libminifi/src/io/BaseStream.cpp b/extensions/script/lua/LuaOutputStream.cpp
similarity index 66%
rename from libminifi/src/io/BaseStream.cpp
rename to extensions/script/lua/LuaOutputStream.cpp
index 013747376..92e91ee16 100644
--- a/libminifi/src/io/BaseStream.cpp
+++ b/extensions/script/lua/LuaOutputStream.cpp
@@ -15,19 +15,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "io/BaseStream.h"
-#include <vector>
+
+#include <memory>
+#include <utility>
#include <string>
-#include "core/expect.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace io {
+#include "LuaOutputStream.h"
+
+namespace org::apache::nifi::minifi::lua {
+
+LuaOutputStream::LuaOutputStream(std::shared_ptr<io::OutputStream> stream)
+ : stream_(std::move(stream)) {
+}
+
+size_t LuaOutputStream::write(std::string buf) {
+ return stream_->write(reinterpret_cast<const uint8_t*>(buf.data()), buf.length());
+}
-} /* namespace io */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/extensions/script/lua/LuaBaseStream.h b/extensions/script/lua/LuaOutputStream.h
similarity index 67%
rename from extensions/script/lua/LuaBaseStream.h
rename to extensions/script/lua/LuaOutputStream.h
index 47b5017df..8bd8c49a6 100644
--- a/extensions/script/lua/LuaBaseStream.h
+++ b/extensions/script/lua/LuaOutputStream.h
@@ -21,23 +21,13 @@
#include <string>
#include "sol/sol.hpp"
-#include "io/BaseStream.h"
+#include "io/OutputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace lua {
+namespace org::apache::nifi::minifi::lua {
-class LuaBaseStream {
+class LuaOutputStream {
public:
- explicit LuaBaseStream(std::shared_ptr<io::BaseStream> stream);
-
- /**
- * Read n bytes of data (returns string, to follow Lua idioms)
- * @return
- */
- std::string read(size_t len = 0);
+ explicit LuaOutputStream(std::shared_ptr<io::OutputStream> stream);
/**
* Write data (receives string, to follow Lua idioms)
@@ -47,11 +37,7 @@ class LuaBaseStream {
size_t write(std::string buf);
private:
- std::shared_ptr<io::BaseStream> stream_;
+ std::shared_ptr<io::OutputStream> stream_;
};
-} /* namespace lua */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/extensions/script/lua/LuaProcessSession.cpp b/extensions/script/lua/LuaProcessSession.cpp
index 7b9916b8c..44a683711 100644
--- a/extensions/script/lua/LuaProcessSession.cpp
+++ b/extensions/script/lua/LuaProcessSession.cpp
@@ -20,11 +20,7 @@
#include "LuaProcessSession.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace lua {
+namespace org::apache::nifi::minifi::lua {
LuaProcessSession::LuaProcessSession(std::shared_ptr<core::ProcessSession> session)
: session_(std::move(session)) {
@@ -48,7 +44,7 @@ std::shared_ptr<script::ScriptFlowFile> LuaProcessSession::get() {
}
void LuaProcessSession::transfer(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file,
- core::Relationship relationship) {
+ const core::Relationship& relationship) {
if (!session_) {
throw std::runtime_error("Access of ProcessSession after it has been released");
}
@@ -74,9 +70,9 @@ void LuaProcessSession::read(const std::shared_ptr<script::ScriptFlowFile> &scri
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->read(flow_file, [&input_stream_callback](const std::shared_ptr<io::BaseStream>& input_stream) -> int64_t {
+ session_->read(flow_file, [&input_stream_callback](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
sol::function callback = input_stream_callback["process"];
- return callback(input_stream_callback, std::make_shared<LuaBaseStream>(input_stream));
+ return callback(input_stream_callback, std::make_shared<LuaInputStream>(input_stream));
});
}
@@ -92,9 +88,9 @@ void LuaProcessSession::write(const std::shared_ptr<script::ScriptFlowFile> &scr
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->write(flow_file, [&output_stream_callback](const std::shared_ptr<io::BaseStream>& output_stream) -> int64_t {
+ session_->write(flow_file, [&output_stream_callback](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
sol::function callback = output_stream_callback["process"];
- return callback(output_stream_callback, std::make_shared<LuaBaseStream>(output_stream));
+ return callback(output_stream_callback, std::make_shared<LuaOutputStream>(output_stream));
});
}
@@ -135,8 +131,4 @@ void LuaProcessSession::releaseCoreResources() {
session_.reset();
}
-} /* namespace lua */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/extensions/script/lua/LuaProcessSession.h b/extensions/script/lua/LuaProcessSession.h
index 27c4709e1..c428239fb 100644
--- a/extensions/script/lua/LuaProcessSession.h
+++ b/extensions/script/lua/LuaProcessSession.h
@@ -24,13 +24,10 @@
#include "../ScriptFlowFile.h"
#include "sol/sol.hpp"
-#include "LuaBaseStream.h"
+#include "LuaInputStream.h"
+#include "LuaOutputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace lua {
+namespace org::apache::nifi::minifi::lua {
class LuaProcessSession {
public:
@@ -39,7 +36,7 @@ class LuaProcessSession {
std::shared_ptr<script::ScriptFlowFile> get();
std::shared_ptr<script::ScriptFlowFile> create();
std::shared_ptr<script::ScriptFlowFile> create(const std::shared_ptr<script::ScriptFlowFile> &flow_file);
- void transfer(const std::shared_ptr<script::ScriptFlowFile> &flow_file, core::Relationship relationship);
+ void transfer(const std::shared_ptr<script::ScriptFlowFile> &flow_file, const core::Relationship& relationship);
void read(const std::shared_ptr<script::ScriptFlowFile> &script_flow_file, sol::table input_stream_callback);
void write(const std::shared_ptr<script::ScriptFlowFile> &flow_file, sol::table output_stream_callback);
@@ -58,8 +55,4 @@ class LuaProcessSession {
std::shared_ptr<core::ProcessSession> session_;
};
-} /* namespace lua */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi::lua
diff --git a/extensions/script/lua/LuaScriptEngine.cpp b/extensions/script/lua/LuaScriptEngine.cpp
index 28142c760..309a89d80 100644
--- a/extensions/script/lua/LuaScriptEngine.cpp
+++ b/extensions/script/lua/LuaScriptEngine.cpp
@@ -54,10 +54,12 @@ LuaScriptEngine::LuaScriptEngine() {
"removeAttribute", &script::ScriptFlowFile::removeAttribute,
"updateAttribute", &script::ScriptFlowFile::updateAttribute,
"setAttribute", &script::ScriptFlowFile::setAttribute);
- lua_.new_usertype<lua::LuaBaseStream>(
- "BaseStream",
- "read", &lua::LuaBaseStream::read,
- "write", &lua::LuaBaseStream::write);
+ lua_.new_usertype<lua::LuaInputStream>(
+ "InputStream",
+ "read", &lua::LuaInputStream::read);
+ lua_.new_usertype<lua::LuaOutputStream>(
+ "OutputStream",
+ "write", &lua::LuaOutputStream::write);
}
void LuaScriptEngine::executeScriptWithAppendedModulePaths(std::string& script) {
diff --git a/extensions/script/python/PyBaseStream.cpp b/extensions/script/python/PyInputStream.cpp
similarity index 80%
rename from extensions/script/python/PyBaseStream.cpp
rename to extensions/script/python/PyInputStream.cpp
index bc51c1ec5..1bc74ceb3 100644
--- a/extensions/script/python/PyBaseStream.cpp
+++ b/extensions/script/python/PyInputStream.cpp
@@ -21,21 +21,21 @@
#include <string>
#include <vector>
-#include "PyBaseStream.h"
+#include "PyInputStream.h"
#include "utils/gsl.h"
namespace org::apache::nifi::minifi::python {
-PyBaseStream::PyBaseStream(std::shared_ptr<io::BaseStream> stream)
+PyInputStream::PyInputStream(std::shared_ptr<io::InputStream> stream)
: stream_(std::move(stream)) {
}
-py::bytes PyBaseStream::read() {
+py::bytes PyInputStream::read() {
return read(stream_->size());
}
-py::bytes PyBaseStream::read(size_t len) {
+py::bytes PyInputStream::read(size_t len) {
if (len == 0) {
len = stream_->size();
}
@@ -50,8 +50,4 @@ py::bytes PyBaseStream::read(size_t len) {
return {reinterpret_cast<char *>(buffer.data()), read};
}
-size_t PyBaseStream::write(const py::bytes& buf) {
- return stream_->write(gsl::make_span(static_cast<std::string>(buf)).as_span<const std::byte>());
-}
-
} // namespace org::apache::nifi::minifi::python
diff --git a/extensions/script/python/PyBaseStream.h b/extensions/script/python/PyInputStream.h
similarity index 84%
copy from extensions/script/python/PyBaseStream.h
copy to extensions/script/python/PyInputStream.h
index ac8bc3dca..d6265a8dc 100644
--- a/extensions/script/python/PyBaseStream.h
+++ b/extensions/script/python/PyInputStream.h
@@ -21,22 +21,21 @@
#include <memory>
#include "pybind11/embed.h"
-#include "io/BaseStream.h"
+#include "io/InputStream.h"
namespace org::apache::nifi::minifi::python {
namespace py = pybind11;
-class PyBaseStream {
+class PyInputStream {
public:
- explicit PyBaseStream(std::shared_ptr<io::BaseStream> stream);
+ explicit PyInputStream(std::shared_ptr<io::InputStream> stream);
py::bytes read();
py::bytes read(size_t len = 0);
- size_t write(const py::bytes& buf);
private:
- std::shared_ptr<io::BaseStream> stream_;
+ std::shared_ptr<io::InputStream> stream_;
};
} // namespace org::apache::nifi::minifi::python
diff --git a/extensions/script/python/PyBaseStream.h b/extensions/script/python/PyOutputStream.cpp
similarity index 72%
copy from extensions/script/python/PyBaseStream.h
copy to extensions/script/python/PyOutputStream.cpp
index ac8bc3dca..8f357ed76 100644
--- a/extensions/script/python/PyBaseStream.h
+++ b/extensions/script/python/PyOutputStream.cpp
@@ -16,27 +16,22 @@
* limitations under the License.
*/
-#pragma once
-
#include <memory>
+#include <utility>
+#include <string>
-#include "pybind11/embed.h"
-#include "io/BaseStream.h"
-
-namespace org::apache::nifi::minifi::python {
+#include "PyOutputStream.h"
-namespace py = pybind11;
+#include "utils/gsl.h"
-class PyBaseStream {
- public:
- explicit PyBaseStream(std::shared_ptr<io::BaseStream> stream);
+namespace org::apache::nifi::minifi::python {
- py::bytes read();
- py::bytes read(size_t len = 0);
- size_t write(const py::bytes& buf);
+PyOutputStream::PyOutputStream(std::shared_ptr<io::OutputStream> stream)
+ : stream_(std::move(stream)) {
+}
- private:
- std::shared_ptr<io::BaseStream> stream_;
-};
+size_t PyOutputStream::write(const py::bytes& buf) {
+ return stream_->write(gsl::make_span(static_cast<std::string>(buf)).as_span<const std::byte>());
+}
} // namespace org::apache::nifi::minifi::python
diff --git a/extensions/script/python/PyBaseStream.h b/extensions/script/python/PyOutputStream.h
similarity index 83%
rename from extensions/script/python/PyBaseStream.h
rename to extensions/script/python/PyOutputStream.h
index ac8bc3dca..c983aa6e2 100644
--- a/extensions/script/python/PyBaseStream.h
+++ b/extensions/script/python/PyOutputStream.h
@@ -21,22 +21,20 @@
#include <memory>
#include "pybind11/embed.h"
-#include "io/BaseStream.h"
+#include "io/OutputStream.h"
namespace org::apache::nifi::minifi::python {
namespace py = pybind11;
-class PyBaseStream {
+class PyOutputStream {
public:
- explicit PyBaseStream(std::shared_ptr<io::BaseStream> stream);
+ explicit PyOutputStream(std::shared_ptr<io::OutputStream> stream);
- py::bytes read();
- py::bytes read(size_t len = 0);
size_t write(const py::bytes& buf);
private:
- std::shared_ptr<io::BaseStream> stream_;
+ std::shared_ptr<io::OutputStream> stream_;
};
} // namespace org::apache::nifi::minifi::python
diff --git a/extensions/script/python/PyProcessSession.cpp b/extensions/script/python/PyProcessSession.cpp
index d9047e0e0..843d00323 100644
--- a/extensions/script/python/PyProcessSession.cpp
+++ b/extensions/script/python/PyProcessSession.cpp
@@ -76,8 +76,8 @@ void PyProcessSession::read(const std::shared_ptr<script::ScriptFlowFile>& scrip
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->read(flow_file, [&input_stream_callback](const std::shared_ptr<io::BaseStream>& input_stream) -> int64_t {
- return input_stream_callback.attr("process")(std::make_shared<PyBaseStream>(input_stream)).cast<int64_t>();
+ session_->read(flow_file, [&input_stream_callback](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
+ return input_stream_callback.attr("process")(std::make_shared<PyInputStream>(input_stream)).cast<int64_t>();
});
}
@@ -93,8 +93,8 @@ void PyProcessSession::write(const std::shared_ptr<script::ScriptFlowFile>& scri
throw std::runtime_error("Access of FlowFile after it has been released");
}
- session_->write(flow_file, [&output_stream_callback](const std::shared_ptr<io::BaseStream>& output_stream) -> int64_t {
- return output_stream_callback.attr("process")(std::make_shared<PyBaseStream>(output_stream)).cast<int64_t>();
+ session_->write(flow_file, [&output_stream_callback](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
+ return output_stream_callback.attr("process")(std::make_shared<PyOutputStream>(output_stream)).cast<int64_t>();
});
}
diff --git a/extensions/script/python/PyProcessSession.h b/extensions/script/python/PyProcessSession.h
index 2aa5473da..898719854 100644
--- a/extensions/script/python/PyProcessSession.h
+++ b/extensions/script/python/PyProcessSession.h
@@ -25,7 +25,8 @@
#include "pybind11/embed.h"
#include "core/ProcessSession.h"
#include "../ScriptFlowFile.h"
-#include "PyBaseStream.h"
+#include "PyInputStream.h"
+#include "PyOutputStream.h"
#if defined(__GNUC__) || defined(__GNUG__)
#pragma GCC visibility push(hidden)
diff --git a/extensions/script/python/PythonBindings.h b/extensions/script/python/PythonBindings.h
index 15f68b3c7..a406f02f8 100644
--- a/extensions/script/python/PythonBindings.h
+++ b/extensions/script/python/PythonBindings.h
@@ -30,7 +30,8 @@
#include "PyProcessSession.h"
#include "PythonProcessor.h"
-#include "PyBaseStream.h"
+#include "PyInputStream.h"
+#include "PyOutputStream.h"
PYBIND11_EMBEDDED_MODULE(minifi_native, m) { // NOLINT
namespace py = pybind11;
@@ -74,8 +75,10 @@ PYBIND11_EMBEDDED_MODULE(minifi_native, m) { // NOLINT
.def("getName", &org::apache::nifi::minifi::core::Relationship::getName)
.def("getDescription", &org::apache::nifi::minifi::core::Relationship::getDescription);
- py::class_<python::PyBaseStream, std::shared_ptr<python::PyBaseStream>>(m, "BaseStream")
- .def("read", static_cast<py::bytes (python::PyBaseStream::*)()>(&python::PyBaseStream::read))
- .def("read", static_cast<py::bytes (python::PyBaseStream::*)(size_t)>(&python::PyBaseStream::read))
- .def("write", &python::PyBaseStream::write);
+ py::class_<python::PyInputStream, std::shared_ptr<python::PyInputStream>>(m, "InputStream")
+ .def("read", static_cast<py::bytes (python::PyInputStream::*)()>(&python::PyInputStream::read))
+ .def("read", static_cast<py::bytes (python::PyInputStream::*)(size_t)>(&python::PyInputStream::read));
+
+ py::class_<python::PyOutputStream, std::shared_ptr<python::PyOutputStream>>(m, "OutputStream")
+ .def("write", &python::PyOutputStream::write);
}
diff --git a/extensions/sftp/client/SFTPClient.cpp b/extensions/sftp/client/SFTPClient.cpp
index 7afcc98df..e2167bfca 100644
--- a/extensions/sftp/client/SFTPClient.cpp
+++ b/extensions/sftp/client/SFTPClient.cpp
@@ -464,7 +464,7 @@ SFTPError SFTPClient::getLastError() const {
return last_error_;
}
-bool SFTPClient::getFile(const std::string& path, io::BaseStream& output, int64_t expected_size /*= -1*/) {
+std::optional<uint64_t> SFTPClient::getFile(const std::string& path, io::OutputStream& output, int64_t expected_size /*= -1*/) {
/**
* SFTP servers should not set the mode of an existing file on open
* (see https://tools.ietf.org/html/draft-ietf-secsh-filexfer-13, Page 33
@@ -493,7 +493,7 @@ bool SFTPClient::getFile(const std::string& path, io::BaseStream& output, int64_
libssh2_session_last_error(ssh_session_, &err_msg, nullptr, 0);
logger_->log_error("Failed to open remote file \"%s\" due to an underlying SSH error: %s", path.c_str(), err_msg);
}
- return false;
+ return std::nullopt;
}
const auto guard = gsl::finally([&file_handle]() {
libssh2_sftp_close(file_handle);
@@ -507,7 +507,7 @@ bool SFTPClient::getFile(const std::string& path, io::BaseStream& output, int64_
if (read_ret < 0) {
last_error_.setSftpError(SFTPError::IoError);
logger_->log_error("Failed to read remote file \"%s\"", path.c_str());
- return false;
+ return std::nullopt;
} else if (read_ret == 0) {
logger_->log_trace("EOF while reading remote file \"%s\"", path.c_str());
break;
@@ -520,7 +520,7 @@ bool SFTPClient::getFile(const std::string& path, io::BaseStream& output, int64_
if (io::isError(write_ret)) {
last_error_.setLibssh2Error(LIBSSH2_FX_OK);
logger_->log_error("Failed to write output");
- return false;
+ return std::nullopt;
}
remaining -= gsl::narrow<decltype(remaining)>(write_ret);
}
@@ -529,13 +529,13 @@ bool SFTPClient::getFile(const std::string& path, io::BaseStream& output, int64_
if (expected_size >= 0 && total_read != gsl::narrow<uint64_t>(expected_size)) {
last_error_.setLibssh2Error(LIBSSH2_FX_OK);
logger_->log_error("Remote file \"%s\" has unexpected size, expected: %ld, actual: %lu", path.c_str(), expected_size, total_read);
- return false;
+ return std::nullopt;
}
- return true;
+ return total_read;
}
-bool SFTPClient::putFile(const std::string& path, io::BaseStream& input, bool overwrite, int64_t expected_size /*= -1*/) {
+std::optional<uint64_t> SFTPClient::putFile(const std::string& path, io::InputStream& input, bool overwrite, int64_t expected_size /*= -1*/) {
int flags = LIBSSH2_FXF_WRITE | LIBSSH2_FXF_CREAT | (overwrite ? LIBSSH2_FXF_TRUNC : LIBSSH2_FXF_EXCL);
logger_->log_trace("Opening remote file \"%s\"", path.c_str());
LIBSSH2_SFTP_HANDLE *file_handle = libssh2_sftp_open(sftp_session_, path.c_str(), flags, 0644);
@@ -559,7 +559,7 @@ bool SFTPClient::putFile(const std::string& path, io::BaseStream& input, bool ov
/* If they just want a zero byte file, we are done */
if (expected_size == 0) {
- return true;
+ return 0;
}
const size_t buf_size = expected_size < 0 ? MAX_BUFFER_SIZE : std::min(gsl::narrow<size_t>(expected_size), MAX_BUFFER_SIZE);
@@ -570,7 +570,7 @@ bool SFTPClient::putFile(const std::string& path, io::BaseStream& input, bool ov
if (io::isError(read_ret)) {
last_error_.setLibssh2Error(LIBSSH2_FX_OK);
logger_->log_error("Error while reading input");
- return false;
+ return std::nullopt;
} else if (read_ret == 0) {
logger_->log_trace("EOF while reading input");
break;
@@ -583,7 +583,7 @@ bool SFTPClient::putFile(const std::string& path, io::BaseStream& input, bool ov
if (write_ret < 0) {
last_error_.setSftpError(SFTPError::IoError);
logger_->log_error("Failed to write remote file \"%s\"", path.c_str());
- return false;
+ return std::nullopt;
}
logger_->log_trace("Wrote %d bytes to remote file \"%s\"", write_ret, path.c_str());
remaining -= gsl::narrow<size_t>(write_ret);
@@ -593,10 +593,10 @@ bool SFTPClient::putFile(const std::string& path, io::BaseStream& input, bool ov
if (expected_size >= 0 && total_read != gsl::narrow<size_t>(expected_size)) {
last_error_.setLibssh2Error(LIBSSH2_FX_OK);
logger_->log_error("Input has unexpected size, expected: %ld, actual: %lu", path.c_str(), expected_size, total_read);
- return false;
+ return std::nullopt;
}
- return true;
+ return total_read;
}
bool SFTPClient::rename(const std::string& source_path, const std::string& target_path, bool overwrite) {
diff --git a/extensions/sftp/client/SFTPClient.h b/extensions/sftp/client/SFTPClient.h
index 9cc48b0ea..8cdf77eee 100644
--- a/extensions/sftp/client/SFTPClient.h
+++ b/extensions/sftp/client/SFTPClient.h
@@ -25,12 +25,12 @@
#include <string>
#include <tuple>
#include <memory>
+#include <optional>
#include "Exception.h"
#include "utils/Enum.h"
#include "utils/BaseHTTPClient.h"
#include "core/logging/Logger.h"
-#include "io/BaseStream.h"
namespace org::apache::nifi::minifi::utils {
@@ -117,9 +117,9 @@ class SFTPClient {
*/
[[nodiscard]] SFTPError getLastError() const;
- bool getFile(const std::string& path, io::BaseStream& output, int64_t expected_size = -1);
+ std::optional<uint64_t> getFile(const std::string& path, io::OutputStream& output, int64_t expected_size = -1);
- bool putFile(const std::string& path, io::BaseStream& input, bool overwrite, int64_t expected_size = -1);
+ std::optional<uint64_t> putFile(const std::string& path, io::InputStream& input, bool overwrite, int64_t expected_size = -1);
bool rename(const std::string& source_path, const std::string& target_path, bool overwrite);
diff --git a/extensions/sftp/processors/FetchSFTP.cpp b/extensions/sftp/processors/FetchSFTP.cpp
index b65a4fffe..c1661249a 100644
--- a/extensions/sftp/processors/FetchSFTP.cpp
+++ b/extensions/sftp/processors/FetchSFTP.cpp
@@ -122,11 +122,12 @@ void FetchSFTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
/* Download file */
try {
- session->write(flow_file, [&remote_file, &client](const std::shared_ptr<io::BaseStream>& stream) -> int64_t {
- if (!client->getFile(remote_file, *stream)) {
+ session->write(flow_file, [&remote_file, &client](const std::shared_ptr<io::OutputStream>& stream) -> int64_t {
+ auto bytes_read = client->getFile(remote_file, *stream);
+ if (!bytes_read) {
throw utils::SFTPException{client->getLastError()};
}
- return gsl::narrow<int64_t>(stream->size());
+ return gsl::narrow<int64_t>(*bytes_read);
});
} catch (const utils::SFTPException& ex) {
logger_->log_debug(ex.what());
diff --git a/extensions/sftp/processors/PutSFTP.cpp b/extensions/sftp/processors/PutSFTP.cpp
index e01eaec81..f47d29e99 100644
--- a/extensions/sftp/processors/PutSFTP.cpp
+++ b/extensions/sftp/processors/PutSFTP.cpp
@@ -290,7 +290,7 @@ bool PutSFTP::processOne(const std::shared_ptr<core::ProcessContext> &context, c
logger_->log_debug("The target path is %s, final target path is %s", target_path.c_str(), final_target_path.c_str());
try {
- session->read(flow_file, [&client, &target_path, this](const std::shared_ptr<io::BaseStream>& stream) {
+ session->read(flow_file, [&client, &target_path, this](const std::shared_ptr<io::InputStream>& stream) {
if (!client->putFile(target_path,
*stream,
conflict_resolution_ == CONFLICT_RESOLUTION_REPLACE /*overwrite*/,
diff --git a/extensions/standard-processors/processors/ExecuteProcess.h b/extensions/standard-processors/processors/ExecuteProcess.h
index 8c735e849..5556c3181 100644
--- a/extensions/standard-processors/processors/ExecuteProcess.h
+++ b/extensions/standard-processors/processors/ExecuteProcess.h
@@ -40,7 +40,6 @@
#include "core/Processor.h"
#include "core/ProcessSession.h"
#include "FlowFileRecord.h"
-#include "io/BaseStream.h"
#include "utils/gsl.h"
namespace org {
diff --git a/extensions/standard-processors/processors/ExtractText.cpp b/extensions/standard-processors/processors/ExtractText.cpp
index 68450057b..bb5d8a46b 100644
--- a/extensions/standard-processors/processors/ExtractText.cpp
+++ b/extensions/standard-processors/processors/ExtractText.cpp
@@ -94,7 +94,7 @@ void ExtractText::onTrigger(core::ProcessContext *context, core::ProcessSession
session->transfer(flowFile, Success);
}
-int64_t ExtractText::ReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+int64_t ExtractText::ReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) const {
size_t read_size = 0;
bool regex_mode;
size_t size_limit = flowFile_->getSize();
diff --git a/extensions/standard-processors/processors/ExtractText.h b/extensions/standard-processors/processors/ExtractText.h
index 0d06a3c53..415093008 100644
--- a/extensions/standard-processors/processors/ExtractText.h
+++ b/extensions/standard-processors/processors/ExtractText.h
@@ -75,7 +75,7 @@ class ExtractText : public core::Processor {
class ReadCallback {
public:
ReadCallback(std::shared_ptr<core::FlowFile> flowFile, core::ProcessContext *ct, std::shared_ptr<core::logging::Logger> lgr);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const;
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) const;
private:
std::shared_ptr<core::FlowFile> flowFile_;
diff --git a/extensions/standard-processors/processors/HashContent.cpp b/extensions/standard-processors/processors/HashContent.cpp
index 2f7379f5f..3c6ecfb79 100644
--- a/extensions/standard-processors/processors/HashContent.cpp
+++ b/extensions/standard-processors/processors/HashContent.cpp
@@ -32,11 +32,7 @@
#include "core/FlowFile.h"
#include "core/Resource.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace processors {
+namespace org::apache::nifi::minifi::processors {
const core::Property HashContent::HashAttribute("Hash Attribute", "Attribute to store checksum to", "Checksum");
const core::Property HashContent::HashAlgorithm("Hash Algorithm", "Name of the algorithm used to generate checksum", "SHA256");
@@ -83,7 +79,7 @@ void HashContent::onTrigger(core::ProcessContext *, core::ProcessSession *sessio
}
logger_->log_trace("attempting read");
- session->read(flowFile, [&flowFile, this](const std::shared_ptr<io::BaseStream>& stream) {
+ session->read(flowFile, [&flowFile, this](const std::shared_ptr<io::InputStream>& stream) {
// This throws in case algo is not found, but that's fine
logger_->log_trace("Searching for %s", algoName_);
auto algo = HashAlgos.at(algoName_);
@@ -99,10 +95,6 @@ void HashContent::onTrigger(core::ProcessContext *, core::ProcessSession *sessio
REGISTER_RESOURCE(HashContent, Processor);
-} // namespace processors
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::processors
#endif // OPENSSL_SUPPORT
diff --git a/extensions/standard-processors/processors/HashContent.h b/extensions/standard-processors/processors/HashContent.h
index fa1fffe84..a0c877c90 100644
--- a/extensions/standard-processors/processors/HashContent.h
+++ b/extensions/standard-processors/processors/HashContent.h
@@ -17,8 +17,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_HASHCONTENT_H_
-#define EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_HASHCONTENT_H_
+#pragma once
#ifdef OPENSSL_SUPPORT
@@ -37,7 +36,6 @@
#include "FlowFileRecord.h"
#include "core/Processor.h"
#include "core/ProcessSession.h"
-#include "io/BaseStream.h"
#include "utils/StringUtils.h"
#include "utils/Export.h"
@@ -47,7 +45,7 @@ using HashReturnType = std::pair<std::string, int64_t>;
namespace { // NOLINT
#define HASH_BUFFER_SIZE 16384
- HashReturnType MD5Hash(const std::shared_ptr<org::apache::nifi::minifi::io::BaseStream>& stream) {
+ HashReturnType MD5Hash(const std::shared_ptr<org::apache::nifi::minifi::io::InputStream>& stream) {
HashReturnType ret_val;
ret_val.second = 0;
std::array<std::byte, HASH_BUFFER_SIZE> buffer{};
@@ -71,7 +69,7 @@ namespace { // NOLINT
return ret_val;
}
- HashReturnType SHA1Hash(const std::shared_ptr<org::apache::nifi::minifi::io::BaseStream>& stream) {
+ HashReturnType SHA1Hash(const std::shared_ptr<org::apache::nifi::minifi::io::InputStream>& stream) {
HashReturnType ret_val;
ret_val.second = 0;
std::array<std::byte, HASH_BUFFER_SIZE> buffer{};
@@ -95,7 +93,7 @@ namespace { // NOLINT
return ret_val;
}
- HashReturnType SHA256Hash(const std::shared_ptr<org::apache::nifi::minifi::io::BaseStream>& stream) {
+ HashReturnType SHA256Hash(const std::shared_ptr<org::apache::nifi::minifi::io::InputStream>& stream) {
HashReturnType ret_val;
ret_val.second = 0;
std::array<std::byte, HASH_BUFFER_SIZE> buffer{};
@@ -123,7 +121,7 @@ namespace { // NOLINT
namespace org::apache::nifi::minifi::processors {
-static const std::map<std::string, const std::function<HashReturnType(const std::shared_ptr<io::BaseStream>&)>> HashAlgos =
+static const std::map<std::string, const std::function<HashReturnType(const std::shared_ptr<io::InputStream>&)>> HashAlgos =
{ {"MD5", MD5Hash}, {"SHA1", SHA1Hash}, {"SHA256", SHA256Hash} };
class HashContent : public core::Processor {
@@ -171,5 +169,3 @@ class HashContent : public core::Processor {
} // namespace org::apache::nifi::minifi::processors
#endif // OPENSSL_SUPPORT
-
-#endif // EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_HASHCONTENT_H_
diff --git a/extensions/standard-processors/processors/PutFile.cpp b/extensions/standard-processors/processors/PutFile.cpp
index caa58d235..baddbb64d 100644
--- a/extensions/standard-processors/processors/PutFile.cpp
+++ b/extensions/standard-processors/processors/PutFile.cpp
@@ -185,7 +185,7 @@ std::string PutFile::tmpWritePath(const std::string &filename, const std::string
return tmpFile;
}
-bool PutFile::putFile(core::ProcessSession *session, std::shared_ptr<core::FlowFile> flowFile, const std::string &tmpFile, const std::string &destFile, const std::string &destDir) {
+bool PutFile::putFile(core::ProcessSession *session, const std::shared_ptr<core::FlowFile>& flowFile, const std::string &tmpFile, const std::string &destFile, const std::string &destDir) {
if (!utils::file::exists(destDir) && try_mkdirs_) {
// Attempt to create directories in file's path
std::stringstream dir_path_stream;
@@ -296,7 +296,7 @@ PutFile::ReadCallback::ReadCallback(std::string tmp_file, std::string dest_file)
}
// Copy the entire file contents to the temporary file
-int64_t PutFile::ReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t PutFile::ReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) {
// Copy file contents into tmp file
write_succeeded_ = false;
size_t size = 0;
diff --git a/extensions/standard-processors/processors/PutFile.h b/extensions/standard-processors/processors/PutFile.h
index ae295dcd4..d2b2e1c11 100644
--- a/extensions/standard-processors/processors/PutFile.h
+++ b/extensions/standard-processors/processors/PutFile.h
@@ -17,8 +17,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_PUTFILE_H_
-#define EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_PUTFILE_H_
+#pragma once
#include <memory>
#include <string>
@@ -32,11 +31,7 @@
#include "utils/Id.h"
#include "utils/Export.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace processors {
+namespace org::apache::nifi::minifi::processors {
class PutFile : public core::Processor {
public:
@@ -92,7 +87,7 @@ class PutFile : public core::Processor {
public:
ReadCallback(std::string tmp_file, std::string dest_file);
~ReadCallback();
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream);
bool commit();
private:
@@ -116,7 +111,7 @@ class PutFile : public core::Processor {
int64_t max_dest_files_ = -1;
bool putFile(core::ProcessSession *session,
- std::shared_ptr<core::FlowFile> flowFile,
+ const std::shared_ptr<core::FlowFile>& flowFile,
const std::string &tmpFile,
const std::string &destFile,
const std::string &destDir);
@@ -140,10 +135,4 @@ class PutFile : public core::Processor {
#endif
};
-} // namespace processors
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
-
-#endif // EXTENSIONS_STANDARD_PROCESSORS_PROCESSORS_PUTFILE_H_
+} // namespace org::apache::nifi::minifi::processors
diff --git a/extensions/standard-processors/processors/RouteText.cpp b/extensions/standard-processors/processors/RouteText.cpp
index 66921d0be..0f1592cb0 100644
--- a/extensions/standard-processors/processors/RouteText.cpp
+++ b/extensions/standard-processors/processors/RouteText.cpp
@@ -130,7 +130,7 @@ class RouteText::ReadCallback {
ReadCallback(Segmentation segmentation, size_t file_size, Fn&& fn)
: segmentation_(segmentation), file_size_(file_size), fn_(std::move(fn)) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) const {
std::vector<std::byte> buffer;
buffer.resize(file_size_);
size_t ret = stream->read(buffer);
diff --git a/extensions/standard-processors/processors/TailFile.cpp b/extensions/standard-processors/processors/TailFile.cpp
index 6c614fd9c..7a3f20194 100644
--- a/extensions/standard-processors/processors/TailFile.cpp
+++ b/extensions/standard-processors/processors/TailFile.cpp
@@ -218,8 +218,8 @@ class FileReaderCallback {
openFile(file_name, offset, input_stream_, logger_);
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& output_stream) {
- io::CRCStream<io::BaseStream> crc_stream{gsl::make_not_null(output_stream.get()), checksum_};
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& output_stream) {
+ io::CRCStream<io::OutputStream> crc_stream{gsl::make_not_null(output_stream.get()), checksum_};
uint64_t num_bytes_written = 0;
bool found_delimiter = false;
@@ -291,10 +291,10 @@ class WholeFileReaderCallback {
return checksum_;
}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& output_stream) {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& output_stream) {
std::array<char, BUFFER_SIZE> buffer;
- io::CRCStream<io::BaseStream> crc_stream{gsl::make_not_null(output_stream.get()), checksum_};
+ io::CRCStream<io::OutputStream> crc_stream{gsl::make_not_null(output_stream.get()), checksum_};
uint64_t num_bytes_written = 0;
diff --git a/extensions/tensorflow/TFApplyGraph.cpp b/extensions/tensorflow/TFApplyGraph.cpp
index 62be2cb65..77e70a6a0 100644
--- a/extensions/tensorflow/TFApplyGraph.cpp
+++ b/extensions/tensorflow/TFApplyGraph.cpp
@@ -182,7 +182,7 @@ void TFApplyGraph::onTrigger(const std::shared_ptr<core::ProcessContext>& /*cont
}
}
-int64_t TFApplyGraph::GraphReadCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFApplyGraph::GraphReadCallback::process(const std::shared_ptr<io::InputStream>& stream) {
std::string graph_proto_buf;
graph_proto_buf.resize(stream->size());
const auto num_read = stream->read(reinterpret_cast<uint8_t *>(&graph_proto_buf[0]), stream->size());
@@ -193,7 +193,7 @@ int64_t TFApplyGraph::GraphReadCallback::process(const std::shared_ptr<io::BaseS
return gsl::narrow<int64_t>(num_read);
}
-int64_t TFApplyGraph::TensorReadCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFApplyGraph::TensorReadCallback::process(const std::shared_ptr<io::InputStream>& stream) {
std::string tensor_proto_buf;
tensor_proto_buf.resize(stream->size());
const auto num_read = stream->read(reinterpret_cast<uint8_t *>(&tensor_proto_buf[0]), stream->size());
@@ -204,7 +204,7 @@ int64_t TFApplyGraph::TensorReadCallback::process(const std::shared_ptr<io::Base
return gsl::narrow<int64_t>(num_read);
}
-int64_t TFApplyGraph::TensorWriteCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFApplyGraph::TensorWriteCallback::process(const std::shared_ptr<io::OutputStream>& stream) {
auto tensor_proto_buf = tensor_proto_->SerializeAsString();
auto num_wrote = stream->write(reinterpret_cast<uint8_t *>(&tensor_proto_buf[0]),
static_cast<int>(tensor_proto_buf.size()));
diff --git a/extensions/tensorflow/TFApplyGraph.h b/extensions/tensorflow/TFApplyGraph.h
index 2a67997d5..edfd5516f 100644
--- a/extensions/tensorflow/TFApplyGraph.h
+++ b/extensions/tensorflow/TFApplyGraph.h
@@ -23,6 +23,8 @@
#include <core/Processor.h>
#include <tensorflow/core/public/session.h>
#include <concurrentqueue.h>
+#include "io/InputStream.h"
+#include "io/OutputStream.h"
namespace org::apache::nifi::minifi::processors {
@@ -83,7 +85,7 @@ class TFApplyGraph : public core::Processor {
: graph_def_(std::move(graph_def)) {
}
~GraphReadCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::InputStream>& stream) override;
private:
std::shared_ptr<tensorflow::GraphDef> graph_def_;
@@ -95,7 +97,7 @@ class TFApplyGraph : public core::Processor {
: tensor_proto_(std::move(tensor_proto)) {
}
~TensorReadCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::InputStream>& stream) override;
private:
std::shared_ptr<tensorflow::TensorProto> tensor_proto_;
@@ -107,7 +109,7 @@ class TFApplyGraph : public core::Processor {
: tensor_proto_(std::move(tensor_proto)) {
}
~TensorWriteCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::OutputStream>& stream) override;
private:
std::shared_ptr<tensorflow::TensorProto> tensor_proto_;
diff --git a/extensions/tensorflow/TFConvertImageToTensor.cpp b/extensions/tensorflow/TFConvertImageToTensor.cpp
index d1bdc5e8f..2e329d928 100644
--- a/extensions/tensorflow/TFConvertImageToTensor.cpp
+++ b/extensions/tensorflow/TFConvertImageToTensor.cpp
@@ -304,7 +304,7 @@ void TFConvertImageToTensor::onTrigger(const std::shared_ptr<core::ProcessContex
}
}
-int64_t TFConvertImageToTensor::ImageReadCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFConvertImageToTensor::ImageReadCallback::process(const std::shared_ptr<io::InputStream>& stream) {
if (tensor_->AllocatedBytes() < stream->size()) {
throw std::runtime_error("Tensor is not big enough to hold FlowFile bytes");
}
@@ -315,7 +315,7 @@ int64_t TFConvertImageToTensor::ImageReadCallback::process(const std::shared_ptr
return num_read;
}
-int64_t TFConvertImageToTensor::TensorWriteCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFConvertImageToTensor::TensorWriteCallback::process(const std::shared_ptr<io::OutputStream>& stream) {
auto tensor_proto_buf = tensor_proto_->SerializeAsString();
auto num_wrote = stream->write(reinterpret_cast<uint8_t *>(&tensor_proto_buf[0]),
static_cast<int>(tensor_proto_buf.size()));
diff --git a/extensions/tensorflow/TFConvertImageToTensor.h b/extensions/tensorflow/TFConvertImageToTensor.h
index ddb33690f..6c71bf180 100644
--- a/extensions/tensorflow/TFConvertImageToTensor.h
+++ b/extensions/tensorflow/TFConvertImageToTensor.h
@@ -23,6 +23,8 @@
#include <core/Processor.h>
#include <tensorflow/core/public/session.h>
#include <concurrentqueue.h>
+#include "io/InputStream.h"
+#include "io/OutputStream.h"
namespace org::apache::nifi::minifi::processors {
@@ -89,7 +91,7 @@ class TFConvertImageToTensor : public core::Processor {
: tensor_(tensor) {
}
~ImageReadCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::InputStream>& stream) override;
private:
tensorflow::Tensor *tensor_;
@@ -101,7 +103,7 @@ class TFConvertImageToTensor : public core::Processor {
: tensor_proto_(std::move(tensor_proto)) {
}
~TensorWriteCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::OutputStream>& stream) override;
private:
std::shared_ptr<tensorflow::TensorProto> tensor_proto_;
diff --git a/extensions/tensorflow/TFExtractTopLabels.cpp b/extensions/tensorflow/TFExtractTopLabels.cpp
index 3e26d6f83..285a08dc4 100644
--- a/extensions/tensorflow/TFExtractTopLabels.cpp
+++ b/extensions/tensorflow/TFExtractTopLabels.cpp
@@ -116,7 +116,7 @@ void TFExtractTopLabels::onTrigger(const std::shared_ptr<core::ProcessContext>&
}
}
-int64_t TFExtractTopLabels::LabelsReadCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFExtractTopLabels::LabelsReadCallback::process(const std::shared_ptr<io::InputStream>& stream) {
size_t total_read = 0;
std::string label;
uint64_t max_label_len = 65536;
@@ -145,7 +145,7 @@ int64_t TFExtractTopLabels::LabelsReadCallback::process(const std::shared_ptr<io
return gsl::narrow<int64_t>(total_read);
}
-int64_t TFExtractTopLabels::TensorReadCallback::process(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t TFExtractTopLabels::TensorReadCallback::process(const std::shared_ptr<io::OutputStream>& stream) {
std::string tensor_proto_buf;
tensor_proto_buf.resize(stream->size());
const auto num_read = stream->read(reinterpret_cast<uint8_t *>(&tensor_proto_buf[0]), stream->size());
diff --git a/extensions/tensorflow/TFExtractTopLabels.h b/extensions/tensorflow/TFExtractTopLabels.h
index cb1446dcb..bd707752d 100644
--- a/extensions/tensorflow/TFExtractTopLabels.h
+++ b/extensions/tensorflow/TFExtractTopLabels.h
@@ -23,6 +23,8 @@
#include <core/Processor.h>
#include <tensorflow/core/public/session.h>
#include <concurrentqueue.h>
+#include "io/InputStream.h"
+#include "io/OutputStream.h"
namespace org::apache::nifi::minifi::processors {
@@ -63,7 +65,7 @@ class TFExtractTopLabels : public core::Processor {
: labels_(std::move(labels)) {
}
~LabelsReadCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::InputStream>& stream) override;
private:
std::shared_ptr<std::vector<std::string>> labels_;
@@ -75,7 +77,7 @@ class TFExtractTopLabels : public core::Processor {
: tensor_proto_(std::move(tensor_proto)) {
}
~TensorReadCallback() override = default;
- int64_t process(const std::shared_ptr<io::BaseStream>& stream) override;
+ int64_t process(const std::shared_ptr<io::OutputStream>& stream) override;
private:
std::shared_ptr<tensorflow::TensorProto> tensor_proto_;
diff --git a/extensions/usb-camera/GetUSBCamera.cpp b/extensions/usb-camera/GetUSBCamera.cpp
index 98e4204e0..72923fc7b 100644
--- a/extensions/usb-camera/GetUSBCamera.cpp
+++ b/extensions/usb-camera/GetUSBCamera.cpp
@@ -384,7 +384,7 @@ GetUSBCamera::PNGWriteCallback::PNGWriteCallback(std::shared_ptr<std::mutex> wri
height_(height) {
}
-int64_t GetUSBCamera::PNGWriteCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t GetUSBCamera::PNGWriteCallback::operator()(const std::shared_ptr<io::OutputStream>& stream) {
std::lock_guard<std::mutex> lock(*png_write_mtx_);
logger_->log_info("Writing %d bytes of raw capture data to PNG output", frame_->data_bytes);
png_structp png = png_create_write_struct(PNG_LIBPNG_VER_STRING, nullptr, nullptr, nullptr);
diff --git a/extensions/usb-camera/GetUSBCamera.h b/extensions/usb-camera/GetUSBCamera.h
index 869529f1f..54a17d2c9 100644
--- a/extensions/usb-camera/GetUSBCamera.h
+++ b/extensions/usb-camera/GetUSBCamera.h
@@ -118,7 +118,7 @@ class GetUSBCamera : public core::Processor {
class PNGWriteCallback {
public:
PNGWriteCallback(std::shared_ptr<std::mutex> write_mtx, uvc_frame_t *frame, uint32_t width, uint32_t height);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream);
private:
std::shared_ptr<std::mutex> png_write_mtx_;
diff --git a/libminifi/include/FlowFileRecord.h b/libminifi/include/FlowFileRecord.h
index 210fa935e..141f3b4ba 100644
--- a/libminifi/include/FlowFileRecord.h
+++ b/libminifi/include/FlowFileRecord.h
@@ -31,7 +31,6 @@
#include <fstream>
#include <set>
#include "core/ContentRepository.h"
-#include "io/BaseStream.h"
#include "core/FlowFile.h"
#include "utils/TimeUtil.h"
#include "core/logging/LoggerFactory.h"
diff --git a/libminifi/include/c2/PayloadSerializer.h b/libminifi/include/c2/PayloadSerializer.h
index effd297e1..aa9dc8bbd 100644
--- a/libminifi/include/c2/PayloadSerializer.h
+++ b/libminifi/include/c2/PayloadSerializer.h
@@ -26,7 +26,8 @@
#include "core/state/Value.h"
#include "c2/C2Protocol.h"
-#include "io/BaseStream.h"
+#include "io/OutputStream.h"
+#include "io/InputStream.h"
#include "io/BufferStream.h"
#include "utils/gsl.h"
@@ -41,7 +42,7 @@ class PayloadSerializer {
/**
* Static function that serializes the value nodes
*/
- static void serializeValueNode(state::response::ValueNode &value, std::shared_ptr<io::BaseStream> stream) {
+ static void serializeValueNode(state::response::ValueNode &value, std::shared_ptr<io::OutputStream> stream) {
auto base_type = value.getValue();
if (!base_type) {
uint8_t type = 0;
@@ -75,7 +76,7 @@ class PayloadSerializer {
stream->write(str);
}
}
- static void serialize(uint16_t op, const C2Payload &payload, std::shared_ptr<io::BaseStream> stream) {
+ static void serialize(uint16_t op, const C2Payload &payload, std::shared_ptr<io::OutputStream> stream) {
uint8_t st;
uint32_t size = gsl::narrow<uint32_t>(payload.getNestedPayloads().size());
stream->write(size);
@@ -143,8 +144,8 @@ class PayloadSerializer {
}
return op;
}
- static std::shared_ptr<io::BaseStream> serialize(uint16_t version, const C2Payload &payload) {
- std::shared_ptr<io::BaseStream> stream = std::make_shared<io::BufferStream>();
+ static std::shared_ptr<io::OutputStream> serialize(uint16_t version, const C2Payload &payload) {
+ std::shared_ptr<io::OutputStream> stream = std::make_shared<io::BufferStream>();
uint16_t op = 0;
uint8_t st = 0;
op = opToInt(payload.getOperation());
@@ -176,7 +177,7 @@ class PayloadSerializer {
return stream;
}
- static state::response::ValueNode deserializeValueNode(io::BaseStream *stream) {
+ static state::response::ValueNode deserializeValueNode(io::InputStream *stream) {
uint8_t type = 0;
stream->read(type);
state::response::ValueNode node;
@@ -224,7 +225,7 @@ class PayloadSerializer {
* @param identifier for this payload
* @param stream base stream in which we will serialize the parent payload.
*/
- static bool deserializePayload(C2Payload &parent, Operation operation, std::string identifier, io::BaseStream *stream) {
+ static bool deserializePayload(C2Payload &parent, Operation operation, std::string identifier, io::InputStream *stream) {
uint32_t payloads = 0;
stream->read(payloads);
uint8_t op{}, st{};
diff --git a/libminifi/include/core/ContentRepository.h b/libminifi/include/core/ContentRepository.h
index e109eb97a..e963d99f5 100644
--- a/libminifi/include/core/ContentRepository.h
+++ b/libminifi/include/core/ContentRepository.h
@@ -25,7 +25,6 @@
#include "properties/Configure.h"
#include "ResourceClaim.h"
#include "io/BufferStream.h"
-#include "io/BaseStream.h"
#include "StreamManager.h"
#include "core/Connectable.h"
#include "ContentSession.h"
diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h
index 29054e008..794d64997 100644
--- a/libminifi/include/core/ProcessSession.h
+++ b/libminifi/include/core/ProcessSession.h
@@ -82,11 +82,11 @@ class ProcessSession : public ReferenceContainer {
// Clone a new UUID FlowFile from parent for attributes and sub set of parent content resource claim
std::shared_ptr<core::FlowFile> clone(const std::shared_ptr<core::FlowFile> &parent, int64_t offset, int64_t size);
// Transfer the FlowFile to the relationship
- virtual void transfer(const std::shared_ptr<core::FlowFile> &flow, Relationship relationship);
+ virtual void transfer(const std::shared_ptr<core::FlowFile> &flow, const Relationship& relationship);
// Put Attribute
- void putAttribute(const std::shared_ptr<core::FlowFile> &flow, std::string key, std::string value);
+ void putAttribute(const std::shared_ptr<core::FlowFile> &flow, const std::string& key, const std::string& value);
// Remove Attribute
- void removeAttribute(const std::shared_ptr<core::FlowFile> &flow, std::string key);
+ void removeAttribute(const std::shared_ptr<core::FlowFile> &flow, const std::string& key);
// Remove Flow File
void remove(const std::shared_ptr<core::FlowFile> &flow);
// Execute the given read callback against the content
diff --git a/libminifi/include/core/ProcessSessionReadCallback.h b/libminifi/include/core/ProcessSessionReadCallback.h
index a89003b90..e4bf7d74d 100644
--- a/libminifi/include/core/ProcessSessionReadCallback.h
+++ b/libminifi/include/core/ProcessSessionReadCallback.h
@@ -17,27 +17,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_CORE_PROCESSSESSIONREADCALLBACK_H_
-#define LIBMINIFI_INCLUDE_CORE_PROCESSSESSIONREADCALLBACK_H_
+#pragma once
#include <memory>
#include <string>
#include "core/logging/LoggerFactory.h"
-#include "io/BaseStream.h"
#include "FlowFileRecord.h"
+#include "io/InputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace core {
+namespace org::apache::nifi::minifi::core {
class ProcessSessionReadCallback {
public:
- ProcessSessionReadCallback(const std::string &tmpFile, const std::string &destFile,
- std::shared_ptr<logging::Logger> logger);
+ ProcessSessionReadCallback(const std::string &tmpFile, std::string destFile, std::shared_ptr<logging::Logger> logger);
~ProcessSessionReadCallback();
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream);
bool commit();
private:
@@ -47,9 +41,4 @@ class ProcessSessionReadCallback {
std::string _tmpFile;
std::string _destFile;
};
-} // namespace core
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
-#endif // LIBMINIFI_INCLUDE_CORE_PROCESSSESSIONREADCALLBACK_H_
+} // namespace org::apache::nifi::minifi::core
diff --git a/libminifi/include/io/BaseStream.h b/libminifi/include/io/BaseStream.h
index bff331f1e..0c7b344a9 100644
--- a/libminifi/include/io/BaseStream.h
+++ b/libminifi/include/io/BaseStream.h
@@ -26,11 +26,7 @@
#include "InputStream.h"
#include "OutputStream.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace io {
+namespace org::apache::nifi::minifi::io {
/**
* Base Stream is the base of a composable stream architecture.
@@ -42,8 +38,4 @@ namespace io {
*/
class BaseStream : public InputStream, public OutputStream {};
-} // namespace io
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::io
diff --git a/libminifi/include/io/CRCStream.h b/libminifi/include/io/CRCStream.h
index 9afd66da0..606f69275 100644
--- a/libminifi/include/io/CRCStream.h
+++ b/libminifi/include/io/CRCStream.h
@@ -33,7 +33,8 @@
#include <arpa/inet.h>
#endif
-#include "BaseStream.h"
+#include "InputStream.h"
+#include "OutputStream.h"
#include "Exception.h"
namespace org {
diff --git a/libminifi/include/io/StreamCallback.h b/libminifi/include/io/StreamCallback.h
index 0116ea746..6d4e6c00a 100644
--- a/libminifi/include/io/StreamCallback.h
+++ b/libminifi/include/io/StreamCallback.h
@@ -21,12 +21,13 @@
namespace org::apache::nifi::minifi::io {
-class BaseStream;
+class InputStream;
+class OutputStream;
// FlowFile IO Callback functions for input and output
// throw exception for error
-using InputStreamCallback = std::function<int64_t(const std::shared_ptr<BaseStream>& input_stream)>;
-using OutputStreamCallback = std::function<int64_t(const std::shared_ptr<BaseStream>& output_stream)>;
-using InputOutputStreamCallback = std::function<int64_t(const std::shared_ptr<BaseStream>& input_stream, const std::shared_ptr<BaseStream>& output_stream)>;
+using InputStreamCallback = std::function<int64_t(const std::shared_ptr<InputStream>& input_stream)>;
+using OutputStreamCallback = std::function<int64_t(const std::shared_ptr<OutputStream>& output_stream)>;
+using InputOutputStreamCallback = std::function<int64_t(const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<OutputStream>& output_stream)>;
} // namespace org::apache::nifi::minifi::io
diff --git a/libminifi/include/io/StreamPipe.h b/libminifi/include/io/StreamPipe.h
index 25364ec1a..c24d04bfa 100644
--- a/libminifi/include/io/StreamPipe.h
+++ b/libminifi/include/io/StreamPipe.h
@@ -23,7 +23,8 @@
#include <functional>
#include <memory>
#include <utility>
-#include "BaseStream.h"
+#include "InputStream.h"
+#include "OutputStream.h"
#include "StreamCallback.h"
namespace org::apache::nifi::minifi {
@@ -62,7 +63,7 @@ class InputStreamPipe {
public:
explicit InputStreamPipe(io::OutputStream& output) : output_(&output) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) const {
return internal::pipe(*stream, *output_);
}
@@ -74,7 +75,7 @@ class OutputStreamPipe {
public:
explicit OutputStreamPipe(io::InputStream& input) : input_(&input) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) const {
return internal::pipe(*input_, *stream);
}
diff --git a/libminifi/include/io/StreamSlice.h b/libminifi/include/io/StreamSlice.h
index aedbd708f..bf53d3a10 100644
--- a/libminifi/include/io/StreamSlice.h
+++ b/libminifi/include/io/StreamSlice.h
@@ -20,7 +20,8 @@
#include <algorithm>
#include <memory>
-#include "BaseStream.h"
+#include "StreamCallback.h"
+#include "InputStream.h"
namespace org::apache::nifi::minifi::io {
@@ -28,17 +29,14 @@ namespace org::apache::nifi::minifi::io {
* A wrapped Base Stream with configurable offset and size
* It hides the original (bigger stream) and acts like the stream starts and ends at the configured offset/size
*/
-class StreamSlice : public BaseStream { // TODO(MINIFICPP-1648) This should be an InputStreamCallback, because writing to Slice is not supported
+class StreamSlice : public InputStream {
public:
- StreamSlice(std::shared_ptr<io::BaseStream>& stream, size_t offset, size_t size);
+ StreamSlice(std::shared_ptr<io::InputStream>& stream, size_t offset, size_t size);
// from InputStream
size_t size() const override { return slice_size_; }
size_t read(gsl::span<std::byte> out_buffer) override;
- // from OutputStream
- size_t write(const uint8_t*, size_t) override { throw std::runtime_error("write is not supported in StreamSlice"); }
-
// from Stream
void close() override { stream_->close(); }
int initialize() override { return stream_->initialize(); }
@@ -48,7 +46,7 @@ class StreamSlice : public BaseStream { // TODO(MINIFICPP-1648) This should be
[[nodiscard]] gsl::span<const std::byte> getBuffer() const override;
private:
- const std::shared_ptr<io::BaseStream>& stream_;
+ const std::shared_ptr<io::InputStream>& stream_;
size_t slice_offset_;
size_t slice_size_;
};
diff --git a/libminifi/include/io/ZlibStream.h b/libminifi/include/io/ZlibStream.h
index 3c1a2d8c9..33097371d 100644
--- a/libminifi/include/io/ZlibStream.h
+++ b/libminifi/include/io/ZlibStream.h
@@ -25,7 +25,7 @@
#include <memory>
#include <vector>
-#include "BaseStream.h"
+#include "OutputStream.h"
#include "core/logging/Logger.h"
#include "utils/gsl.h"
diff --git a/libminifi/include/sitetosite/SiteToSiteClient.h b/libminifi/include/sitetosite/SiteToSiteClient.h
index be1b3b0ab..8d57558f1 100644
--- a/libminifi/include/sitetosite/SiteToSiteClient.h
+++ b/libminifi/include/sitetosite/SiteToSiteClient.h
@@ -16,8 +16,7 @@
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_SITETOSITE_SITETOSITECLIENT_H_
-#define LIBMINIFI_INCLUDE_SITETOSITE_SITETOSITECLIENT_H_
+#pragma once
#include <algorithm>
#include <array>
@@ -34,11 +33,7 @@
#include "core/Connectable.h"
#include "utils/gsl.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace sitetosite {
+namespace org::apache::nifi::minifi::sitetosite {
/**
* Represents a piece of data that is to be sent to or that was received from a
@@ -219,7 +214,7 @@ class SiteToSiteClient : public core::Connectable {
// read Respond
virtual int readResponse(const std::shared_ptr<Transaction> &transaction, RespondCode &code, std::string &message);
// write respond
- virtual int writeResponse(const std::shared_ptr<Transaction> &transaction, RespondCode code, std::string message);
+ virtual int writeResponse(const std::shared_ptr<Transaction> &transaction, RespondCode code, const std::string& message);
// getRespondCodeContext
virtual RespondCodeContext *getRespondCodeContext(RespondCode code) {
for (auto & i : SiteToSiteRequest::respondCodeContext) {
@@ -266,10 +261,4 @@ class SiteToSiteClient : public core::Connectable {
std::shared_ptr<core::logging::Logger> logger_{core::logging::LoggerFactory<SiteToSiteClient>::getLogger()};
};
-} // namespace sitetosite
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
-
-#endif // LIBMINIFI_INCLUDE_SITETOSITE_SITETOSITECLIENT_H_
+} // namespace org::apache::nifi::minifi::sitetosite
diff --git a/libminifi/include/utils/ByteArrayCallback.h b/libminifi/include/utils/ByteArrayCallback.h
index 3f68fa878..9d1fb4c85 100644
--- a/libminifi/include/utils/ByteArrayCallback.h
+++ b/libminifi/include/utils/ByteArrayCallback.h
@@ -14,8 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef LIBMINIFI_INCLUDE_UTILS_BYTEARRAYCALLBACK_H_
-#define LIBMINIFI_INCLUDE_UTILS_BYTEARRAYCALLBACK_H_
+#pragma once
#include <memory>
#include <string>
@@ -36,7 +35,7 @@ class ByteInputCallback {
public:
virtual ~ByteInputCallback() = default;
- virtual int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ virtual int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
stream->seek(0);
if (stream->size() > 0) {
@@ -101,7 +100,7 @@ class ByteOutputCallback {
close();
}
- virtual int64_t operator()(const std::shared_ptr<io::BaseStream>& stream);
+ virtual int64_t operator()(const std::shared_ptr<io::InputStream>& stream);
virtual std::vector<char> to_string();
@@ -141,17 +140,4 @@ class ByteOutputCallback {
std::shared_ptr<core::logging::Logger> logger_;
};
-class StreamOutputCallback : public ByteOutputCallback {
- public:
- explicit StreamOutputCallback(size_t max_size, bool wait_on_read = false)
- : ByteOutputCallback(max_size, wait_on_read) {
- }
-
- void write(char *data, size_t size) override;
-
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) override;
-};
-
} // namespace org::apache::nifi::minifi::utils
-
-#endif // LIBMINIFI_INCLUDE_UTILS_BYTEARRAYCALLBACK_H_
diff --git a/libminifi/include/utils/FileReaderCallback.h b/libminifi/include/utils/FileReaderCallback.h
index e1030a547..3d8233716 100644
--- a/libminifi/include/utils/FileReaderCallback.h
+++ b/libminifi/include/utils/FileReaderCallback.h
@@ -24,11 +24,7 @@
#include "io/StreamPipe.h"
#include "core/logging/Logger.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
/**
* Simple callback to read a file, to be used with ProcessSession::write().
@@ -36,7 +32,7 @@ namespace utils {
class FileReaderCallback {
public:
explicit FileReaderCallback(std::string file_name);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& output_stream) const;
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& output_stream) const;
private:
std::string file_name_;
@@ -49,8 +45,4 @@ class FileReaderCallbackIOError : public std::runtime_error {
int error_code;
};
-} // namespace utils
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/include/utils/JsonCallback.h b/libminifi/include/utils/JsonCallback.h
index 99bb07240..4c37afce0 100644
--- a/libminifi/include/utils/JsonCallback.h
+++ b/libminifi/include/utils/JsonCallback.h
@@ -36,7 +36,7 @@ namespace utils {
class JsonInputCallback {
public:
explicit JsonInputCallback(rapidjson::Document& document) : document_(document) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<io::InputStream>& stream) {
std::string content;
content.resize(stream->size());
const auto read_ret = stream->read(gsl::make_span(content).as_span<std::byte>());
@@ -58,7 +58,7 @@ class JsonOutputCallback {
explicit JsonOutputCallback(rapidjson::Document&& root, std::optional<uint8_t> decimal_places)
: root_(std::move(root)), decimal_places_(decimal_places) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) const {
rapidjson::StringBuffer buffer;
rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
if (decimal_places_.has_value())
@@ -78,7 +78,7 @@ class PrettyJsonOutputCallback {
explicit PrettyJsonOutputCallback(rapidjson::Document&& root, std::optional<uint8_t> decimal_places)
: root_(std::move(root)), decimal_places_(decimal_places) {}
- int64_t operator()(const std::shared_ptr<io::BaseStream>& stream) const {
+ int64_t operator()(const std::shared_ptr<io::OutputStream>& stream) const {
rapidjson::StringBuffer buffer;
rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer);
if (decimal_places_.has_value())
diff --git a/libminifi/include/utils/LineByLineInputOutputStreamCallback.h b/libminifi/include/utils/LineByLineInputOutputStreamCallback.h
index 6538020e2..0d68fe568 100644
--- a/libminifi/include/utils/LineByLineInputOutputStreamCallback.h
+++ b/libminifi/include/utils/LineByLineInputOutputStreamCallback.h
@@ -23,7 +23,8 @@
#include <vector>
#include "core/logging/Logger.h"
-#include "io/BaseStream.h"
+#include "io/InputStream.h"
+#include "io/OutputStream.h"
#include "io/StreamPipe.h"
namespace org::apache::nifi::minifi::utils {
@@ -32,7 +33,7 @@ class LineByLineInputOutputStreamCallback {
public:
using CallbackType = std::function<std::string(const std::string& input_line, bool is_first_line, bool is_last_line)>;
explicit LineByLineInputOutputStreamCallback(CallbackType callback);
- int64_t operator()(const std::shared_ptr<io::BaseStream>& input, const std::shared_ptr<io::BaseStream>& output);
+ int64_t operator()(const std::shared_ptr<io::InputStream>& input, const std::shared_ptr<io::OutputStream>& output);
private:
int64_t readInput(io::InputStream& stream);
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index ff5cbb1c7..7eca66138 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -201,14 +201,14 @@ void ProcessSession::remove(const std::shared_ptr<core::FlowFile> &flow) {
provenance_report_->drop(flow, reason);
}
-void ProcessSession::putAttribute(const std::shared_ptr<core::FlowFile> &flow, std::string key, std::string value) {
+void ProcessSession::putAttribute(const std::shared_ptr<core::FlowFile> &flow, const std::string& key, const std::string& value) {
flow->setAttribute(key, value);
std::stringstream details;
details << process_context_->getProcessorNode()->getName() << " modify flow record " << flow->getUUIDStr() << " attribute " << key << ":" << value;
provenance_report_->modifyAttributes(flow, details.str());
}
-void ProcessSession::removeAttribute(const std::shared_ptr<core::FlowFile> &flow, std::string key) {
+void ProcessSession::removeAttribute(const std::shared_ptr<core::FlowFile> &flow, const std::string& key) {
flow->removeAttribute(key);
std::stringstream details;
details << process_context_->getProcessorNode()->getName() << " remove flow record " << flow->getUUIDStr() << " attribute " + key;
@@ -221,7 +221,7 @@ void ProcessSession::penalize(const std::shared_ptr<core::FlowFile> &flow) {
flow->penalize(penalization_period);
}
-void ProcessSession::transfer(const std::shared_ptr<core::FlowFile> &flow, Relationship relationship) {
+void ProcessSession::transfer(const std::shared_ptr<core::FlowFile> &flow, const Relationship& relationship) {
logging::LOG_INFO(logger_) << "Transferring " << flow->getUUIDStr() << " from " << process_context_->getProcessorNode()->getName() << " to relationship " << relationship.getName();
utils::Identifier uuid = flow->getUUID();
_transferRelationship[uuid] = relationship;
@@ -267,7 +267,7 @@ void ProcessSession::writeBuffer(const std::shared_ptr<core::FlowFile>& flow_fil
writeBuffer(flow_file, buffer.as_span<const std::byte>());
}
void ProcessSession::writeBuffer(const std::shared_ptr<core::FlowFile>& flow_file, gsl::span<const std::byte> buffer) {
- write(flow_file, [buffer](const std::shared_ptr<io::BaseStream>& output_stream) {
+ write(flow_file, [buffer](const std::shared_ptr<io::OutputStream>& output_stream) {
const auto write_status = output_stream->write(buffer);
return io::isError(write_status) ? -1 : gsl::narrow<int64_t>(write_status);
});
@@ -318,7 +318,7 @@ void ProcessSession::appendBuffer(const std::shared_ptr<core::FlowFile>& flow_fi
appendBuffer(flow_file, buffer.as_span<const std::byte>());
}
void ProcessSession::appendBuffer(const std::shared_ptr<core::FlowFile>& flow_file, gsl::span<const std::byte> buffer) {
- append(flow_file, [buffer](const std::shared_ptr<io::BaseStream>& output_stream) {
+ append(flow_file, [buffer](const std::shared_ptr<io::OutputStream>& output_stream) {
const auto write_status = output_stream->write(buffer);
return io::isError(write_status) ? -1 : gsl::narrow<int64_t>(write_status);
});
@@ -339,7 +339,7 @@ int64_t ProcessSession::read(const std::shared_ptr<core::FlowFile> &flow, const
claim = flow->getResourceClaim();
- std::shared_ptr<io::BaseStream> stream = content_session_->read(claim);
+ std::shared_ptr<io::InputStream> stream = content_session_->read(claim);
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile content for read");
@@ -410,7 +410,7 @@ int64_t ProcessSession::readWrite(const std::shared_ptr<core::FlowFile> &flow, c
detail::ReadBufferResult ProcessSession::readBuffer(const std::shared_ptr<core::FlowFile>& flow) {
detail::ReadBufferResult result;
- result.status = read(flow, [&result, this](const std::shared_ptr<io::BaseStream>& input_stream) {
+ result.status = read(flow, [&result, this](const std::shared_ptr<io::InputStream>& input_stream) {
result.buffer.resize(input_stream->size());
const auto read_status = input_stream->read(result.buffer);
if (read_status != result.buffer.size()) {
diff --git a/libminifi/src/core/ProcessSessionReadCallback.cpp b/libminifi/src/core/ProcessSessionReadCallback.cpp
index 810e46c89..caaab46b9 100644
--- a/libminifi/src/core/ProcessSessionReadCallback.cpp
+++ b/libminifi/src/core/ProcessSessionReadCallback.cpp
@@ -24,26 +24,21 @@
#include <string>
#include "core/logging/LoggerConfiguration.h"
-#include "io/BaseStream.h"
#include "utils/gsl.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace core {
+namespace org::apache::nifi::minifi::core {
ProcessSessionReadCallback::ProcessSessionReadCallback(const std::string &tmpFile,
- const std::string &destFile,
+ std::string destFile,
std::shared_ptr<logging::Logger> logger)
- : logger_(logger),
+ : logger_(std::move(logger)),
_tmpFileOs(tmpFile, std::ios::binary),
_tmpFile(tmpFile),
- _destFile(destFile) {
+ _destFile(std::move(destFile)) {
}
// Copy the entire file contents to the temporary file
-int64_t ProcessSessionReadCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t ProcessSessionReadCallback::operator()(const std::shared_ptr<io::InputStream>& stream) {
// Copy file contents into tmp file
_writeSucceeded = false;
size_t size = 0;
@@ -95,8 +90,4 @@ ProcessSessionReadCallback::~ProcessSessionReadCallback() {
std::remove(_tmpFile.c_str());
}
-} // namespace core
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::core
diff --git a/libminifi/src/io/StreamSlice.cpp b/libminifi/src/io/StreamSlice.cpp
index d87f19345..182a6f511 100644
--- a/libminifi/src/io/StreamSlice.cpp
+++ b/libminifi/src/io/StreamSlice.cpp
@@ -20,7 +20,7 @@
namespace org::apache::nifi::minifi::io {
-StreamSlice::StreamSlice(std::shared_ptr<io::BaseStream>& stream, size_t offset, size_t size) : stream_(stream), slice_offset_(offset), slice_size_(size) {
+StreamSlice::StreamSlice(std::shared_ptr<io::InputStream>& stream, size_t offset, size_t size) : stream_(stream), slice_offset_(offset), slice_size_(size) {
stream_->seek(slice_offset_);
if (stream_->size() < slice_offset_ + slice_size_)
throw std::invalid_argument("StreamSlice is bigger than the Stream");
diff --git a/libminifi/src/sitetosite/SiteToSiteClient.cpp b/libminifi/src/sitetosite/SiteToSiteClient.cpp
index e0cfe3670..514599289 100644
--- a/libminifi/src/sitetosite/SiteToSiteClient.cpp
+++ b/libminifi/src/sitetosite/SiteToSiteClient.cpp
@@ -74,7 +74,7 @@ void SiteToSiteClient::deleteTransaction(const utils::Identifier& transactionID)
known_transactions_.erase(transactionID);
}
-int SiteToSiteClient::writeResponse(const std::shared_ptr<Transaction>& /*transaction*/, RespondCode code, std::string message) {
+int SiteToSiteClient::writeResponse(const std::shared_ptr<Transaction>& /*transaction*/, RespondCode code, const std::string& message) {
RespondCodeContext *resCode = this->getRespondCodeContext(code);
if (!resCode) {
return -1;
@@ -465,7 +465,7 @@ int16_t SiteToSiteClient::send(const utils::Identifier &transactionID, DataPacke
return -1;
}
if (flowFile->getSize() > 0) {
- session->read(flowFile, [packet](const std::shared_ptr<io::BaseStream>& input_stream) -> int64_t {
+ session->read(flowFile, [packet](const std::shared_ptr<io::InputStream>& input_stream) -> int64_t {
const auto result = internal::pipe(*input_stream, packet->transaction_->getStream());
if (result == -1) return -1;
packet->_size = gsl::narrow<size_t>(result);
@@ -692,7 +692,7 @@ bool SiteToSiteClient::receiveFlowFiles(const std::shared_ptr<core::ProcessConte
}
if (packet._size > 0) {
- session->write(flowFile, [&packet](const std::shared_ptr<io::BaseStream>& output_stream) -> int64_t {
+ session->write(flowFile, [&packet](const std::shared_ptr<io::OutputStream>& output_stream) -> int64_t {
return internal::pipe(packet.transaction_->getStream(), *output_stream);
});
if (flowFile->getSize() != packet._size) {
diff --git a/libminifi/src/utils/ByteArrayCallback.cpp b/libminifi/src/utils/ByteArrayCallback.cpp
index 8fe16a946..018d24790 100644
--- a/libminifi/src/utils/ByteArrayCallback.cpp
+++ b/libminifi/src/utils/ByteArrayCallback.cpp
@@ -24,13 +24,9 @@
#include "utils/gsl.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
-int64_t ByteOutputCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
+int64_t ByteOutputCallback::operator()(const std::shared_ptr<io::InputStream>& stream) {
stream->seek(0);
if (stream->size() > 0) {
std::vector<std::byte> buffer;
@@ -42,20 +38,6 @@ int64_t ByteOutputCallback::operator()(const std::shared_ptr<io::BaseStream>& st
return gsl::narrow<int64_t>(size_.load());
}
-int64_t StreamOutputCallback::operator()(const std::shared_ptr<io::BaseStream>& stream) {
- stream->seek(0);
- std::unique_ptr<char> buffer = std::unique_ptr<char>(new char[size_.load()]);
- auto written = readFully(buffer.get(), size_);
- stream->write(reinterpret_cast<uint8_t*>(buffer.get()), written);
- return gsl::narrow<int64_t>(stream->size());
-}
-
-void StreamOutputCallback::write(char *data, size_t size) {
- if (!is_alive_)
- return;
- write_and_notify(data, size);
-}
-
std::vector<char> ByteOutputCallback::to_string() {
std::vector<char> buffer;
buffer.resize(size_.load());
@@ -173,8 +155,5 @@ bool ByteOutputCallback::preload_next_str() {
size_ -= current_str.size();
return true;
}
-} /* namespace utils */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/utils/FileReaderCallback.cpp b/libminifi/src/utils/FileReaderCallback.cpp
index 825ae3647..5d6e54c01 100644
--- a/libminifi/src/utils/FileReaderCallback.cpp
+++ b/libminifi/src/utils/FileReaderCallback.cpp
@@ -29,18 +29,14 @@ constexpr std::size_t BUFFER_SIZE = 4096;
} // namespace
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace utils {
+namespace org::apache::nifi::minifi::utils {
FileReaderCallback::FileReaderCallback(std::string file_name)
: file_name_{std::move(file_name)},
logger_(core::logging::LoggerFactory<FileReaderCallback>::getLogger()) {
}
-int64_t FileReaderCallback::operator()(const std::shared_ptr<io::BaseStream>& output_stream) const {
+int64_t FileReaderCallback::operator()(const std::shared_ptr<io::OutputStream>& output_stream) const {
std::array<char, BUFFER_SIZE> buffer;
uint64_t num_bytes_written = 0;
@@ -66,8 +62,4 @@ int64_t FileReaderCallback::operator()(const std::shared_ptr<io::BaseStream>& ou
return num_bytes_written;
}
-} // namespace utils
-} // namespace minifi
-} // namespace nifi
-} // namespace apache
-} // namespace org
+} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/utils/LineByLineInputOutputStreamCallback.cpp b/libminifi/src/utils/LineByLineInputOutputStreamCallback.cpp
index f06ba7786..9f11131d5 100644
--- a/libminifi/src/utils/LineByLineInputOutputStreamCallback.cpp
+++ b/libminifi/src/utils/LineByLineInputOutputStreamCallback.cpp
@@ -25,7 +25,7 @@ LineByLineInputOutputStreamCallback::LineByLineInputOutputStreamCallback(Callbac
: callback_(std::move(callback)) {
}
-int64_t LineByLineInputOutputStreamCallback::operator()(const std::shared_ptr<io::BaseStream>& input, const std::shared_ptr<io::BaseStream>& output) {
+int64_t LineByLineInputOutputStreamCallback::operator()(const std::shared_ptr<io::InputStream>& input, const std::shared_ptr<io::OutputStream>& output) {
gsl_Expects(input);
gsl_Expects(output);
diff --git a/libminifi/test/archive-tests/CompressContentTests.cpp b/libminifi/test/archive-tests/CompressContentTests.cpp
index df2bcf208..f061a08ab 100644
--- a/libminifi/test/archive-tests/CompressContentTests.cpp
+++ b/libminifi/test/archive-tests/CompressContentTests.cpp
@@ -54,7 +54,7 @@ class ReadCallback {
ReadCallback& operator=(const ReadCallback&) = delete;
ReadCallback& operator=(ReadCallback&&) = delete;
- int64_t operator()(const std::shared_ptr<minifi::io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<minifi::io::InputStream>& stream) {
int64_t total_read = 0;
do {
const auto ret = stream->read(gsl::make_span(buffer_).subspan(read_size_));
diff --git a/libminifi/test/archive-tests/MergeFileTests.cpp b/libminifi/test/archive-tests/MergeFileTests.cpp
index dc5e3ba64..d3bb0c1c0 100644
--- a/libminifi/test/archive-tests/MergeFileTests.cpp
+++ b/libminifi/test/archive-tests/MergeFileTests.cpp
@@ -97,7 +97,7 @@ class FixedBuffer {
} while (size_ != capacity_);
return total_read;
}
- int64_t operator()(const std::shared_ptr<minifi::io::BaseStream>& stream) {
+ int64_t operator()(const std::shared_ptr<minifi::io::InputStream>& stream) {
return write(*stream, capacity_);
}
@@ -680,7 +680,7 @@ TEST_CASE_METHOD(MergeTestController, "Test Merge File Attributes Keeping All Un
REQUIRE(attributes["mime.type"] == "application/tar");
}
-void writeString(const std::string& str, const std::shared_ptr<minifi::io::BaseStream>& out) {
+void writeString(const std::string& str, const std::shared_ptr<minifi::io::OutputStream>& out) {
out->write(reinterpret_cast<const uint8_t*>(str.data()), str.length());
}
diff --git a/libminifi/test/rocksdb-tests/ContentSessionTests.cpp b/libminifi/test/rocksdb-tests/ContentSessionTests.cpp
index 04c906a42..a73e9c19e 100644
--- a/libminifi/test/rocksdb-tests/ContentSessionTests.cpp
+++ b/libminifi/test/rocksdb-tests/ContentSessionTests.cpp
@@ -47,12 +47,12 @@ class ContentSessionController : public TestController {
std::shared_ptr<core::ContentRepository> contentRepository;
};
-const std::shared_ptr<minifi::io::BaseStream>& operator<<(const std::shared_ptr<minifi::io::BaseStream>& stream, const std::string& str) {
+const std::shared_ptr<minifi::io::OutputStream>& operator<<(const std::shared_ptr<minifi::io::OutputStream>& stream, const std::string& str) {
REQUIRE(stream->write(reinterpret_cast<const uint8_t*>(str.data()), str.length()) == str.length());
return stream;
}
-const std::shared_ptr<minifi::io::BaseStream>& operator>>(const std::shared_ptr<minifi::io::BaseStream>& stream, std::string& str) {
+const std::shared_ptr<minifi::io::InputStream>& operator>>(const std::shared_ptr<minifi::io::InputStream>& stream, std::string& str) {
str = "";
std::array<std::byte, 4096> buffer{};
while (true) {
diff --git a/libminifi/test/rocksdb-tests/SwapTests.cpp b/libminifi/test/rocksdb-tests/SwapTests.cpp
index 053f08333..97716a58d 100644
--- a/libminifi/test/rocksdb-tests/SwapTests.cpp
+++ b/libminifi/test/rocksdb-tests/SwapTests.cpp
@@ -56,7 +56,7 @@ class OutputProcessor : public core::Processor {
auto id = std::to_string(next_id_++);
auto ff = session->create();
ff->addAttribute("index", id);
- session->write(ff, [&] (const std::shared_ptr<minifi::io::BaseStream>& output) -> int64_t {
+ session->write(ff, [&] (const std::shared_ptr<minifi::io::OutputStream>& output) -> int64_t {
auto ret = output->write(gsl::span<const char>(id.data(), id.size()).as_span<const std::byte>());
if (minifi::io::isError(ret)) {
return -1;
diff --git a/libminifi/test/unit/CRCTests.cpp b/libminifi/test/unit/CRCTests.cpp
index a6d08b45e..3defb7eca 100644
--- a/libminifi/test/unit/CRCTests.cpp
+++ b/libminifi/test/unit/CRCTests.cpp
@@ -19,21 +19,21 @@
#include <string>
#include <vector>
#include "io/CRCStream.h"
-#include "io/BufferStream.h"
+#include "io/OutputStream.h"
#include "../TestBase.h"
#include "../Catch.h"
#include "utils/gsl.h"
TEST_CASE("Test CRC1", "[testcrc1]") {
org::apache::nifi::minifi::io::BufferStream base;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test(gsl::make_not_null(&base));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test(gsl::make_not_null(&base));
test.write(reinterpret_cast<const uint8_t*>("cow"), 3);
REQUIRE(2580823964 == test.getCRC());
}
TEST_CASE("Test CRC2", "[testcrc2]") {
org::apache::nifi::minifi::io::BufferStream base;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test(gsl::make_not_null(&base));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test(gsl::make_not_null(&base));
std::string fox = "the quick brown fox jumped over the brown fox";
std::vector<uint8_t> charvect(fox.begin(), fox.end());
test.write(charvect, charvect.size());
@@ -42,7 +42,7 @@ TEST_CASE("Test CRC2", "[testcrc2]") {
TEST_CASE("Test CRC3", "[testcrc3]") {
org::apache::nifi::minifi::io::BufferStream base;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test(gsl::make_not_null(&base));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test(gsl::make_not_null(&base));
uint64_t number = 7;
test.write(number);
REQUIRE(4215687882 == test.getCRC());
@@ -50,7 +50,7 @@ TEST_CASE("Test CRC3", "[testcrc3]") {
TEST_CASE("Test CRC4", "[testcrc4]") {
org::apache::nifi::minifi::io::BufferStream base;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test(gsl::make_not_null(&base));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test(gsl::make_not_null(&base));
uint32_t number = 7;
test.write(number);
REQUIRE(3206564543 == test.getCRC());
@@ -58,7 +58,7 @@ TEST_CASE("Test CRC4", "[testcrc4]") {
TEST_CASE("Test CRC5", "[testcrc5]") {
org::apache::nifi::minifi::io::BufferStream base;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test(gsl::make_not_null(&base));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test(gsl::make_not_null(&base));
uint16_t number = 7;
test.write(number);
REQUIRE(3753740124 == test.getCRC());
@@ -66,10 +66,10 @@ TEST_CASE("Test CRC5", "[testcrc5]") {
TEST_CASE("CRCStream with initial crc = 0 is the same as without initial crc", "[initial_crc_arg]") {
org::apache::nifi::minifi::io::BufferStream base1;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test_noinit(gsl::make_not_null(&base1));
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test_noinit(gsl::make_not_null(&base1));
org::apache::nifi::minifi::io::BufferStream base2;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test_initzero(gsl::make_not_null(&base2), 0);
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test_initzero(gsl::make_not_null(&base2), 0);
const std::string textString = "The quick brown fox jumps over the lazy dog";
std::vector<uint8_t> textVector1(textString.begin(), textString.end());
@@ -84,17 +84,17 @@ TEST_CASE("CRCStream: one long write is the same as writing in two pieces", "[in
const std::string textString = "The quick brown fox jumps over the lazy dog";
org::apache::nifi::minifi::io::BufferStream base_full;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test_full(gsl::make_not_null(&base_full), 0);
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test_full(gsl::make_not_null(&base_full), 0);
std::vector<uint8_t> textVector_full(textString.begin(), textString.end());
test_full.write(textVector_full, textVector_full.size());
org::apache::nifi::minifi::io::BufferStream base_piece1;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test_piece1(gsl::make_not_null(&base_piece1), 0);
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test_piece1(gsl::make_not_null(&base_piece1), 0);
std::vector<uint8_t> textVector_piece1(textString.begin(), textString.begin() + 15);
test_piece1.write(textVector_piece1, textVector_piece1.size());
org::apache::nifi::minifi::io::BufferStream base_piece2;
- org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::BaseStream> test_piece2(gsl::make_not_null(&base_piece2), test_piece1.getCRC());
+ org::apache::nifi::minifi::io::CRCStream<org::apache::nifi::minifi::io::OutputStream> test_piece2(gsl::make_not_null(&base_piece2), test_piece1.getCRC());
std::vector<uint8_t> textVector_piece2(textString.begin() + 15, textString.end());
test_piece2.write(textVector_piece2, textVector_piece2.size());
diff --git a/libminifi/test/unit/ContentRepositoryDependentTests.h b/libminifi/test/unit/ContentRepositoryDependentTests.h
index 17c60e612..a476b4558 100644
--- a/libminifi/test/unit/ContentRepositoryDependentTests.h
+++ b/libminifi/test/unit/ContentRepositoryDependentTests.h
@@ -36,7 +36,7 @@ namespace ContentRepositoryDependentTests {
struct ReadUntilItCan {
std::string value_;
- int64_t operator()(const std::shared_ptr<minifi::io::BaseStream> &stream) {
+ int64_t operator()(const std::shared_ptr<minifi::io::InputStream> &stream) {
value_.clear();
std::array<std::byte, 1024> buffer{};
size_t bytes_read = 0;
diff --git a/libminifi/test/unit/FlowFileSerializationTests.cpp b/libminifi/test/unit/FlowFileSerializationTests.cpp
index dcfca26eb..b623f4887 100644
--- a/libminifi/test/unit/FlowFileSerializationTests.cpp
+++ b/libminifi/test/unit/FlowFileSerializationTests.cpp
@@ -19,7 +19,6 @@
#include <memory>
#include <string>
-#include "io/BaseStream.h"
#include "serialization/FlowFileV3Serializer.h"
#include "serialization/PayloadSerializer.h"
#include "core/FlowFile.h"
diff --git a/libminifi/test/unit/SerializationTests.cpp b/libminifi/test/unit/SerializationTests.cpp
index 880bec58e..8ba130f92 100644
--- a/libminifi/test/unit/SerializationTests.cpp
+++ b/libminifi/test/unit/SerializationTests.cpp
@@ -19,7 +19,6 @@
#include <memory>
#include <string>
-#include "io/BaseStream.h"
#include "SiteToSiteHelper.h"
#include "../TestBase.h"
#include "../Catch.h"
diff --git a/libminifi/test/unit/StreamTests.cpp b/libminifi/test/unit/StreamTests.cpp
index d2df3f0b3..dc77c5438 100644
--- a/libminifi/test/unit/StreamTests.cpp
+++ b/libminifi/test/unit/StreamTests.cpp
@@ -83,14 +83,16 @@ TEST_CASE("TestWrite1", "[testwrite]") {
TEST_CASE("InvalidStreamSliceTest", "[teststreamslice]") {
std::shared_ptr<minifi::io::BaseStream> base = std::make_shared<minifi::io::BufferStream>();
base->write((const uint8_t*)"\x01\x02\x03\x04\x05\x06\x07\x08", 8);
- REQUIRE_THROWS_WITH(std::make_shared<minifi::io::StreamSlice>(base, 0, 9), "StreamSlice is bigger than the Stream");
- REQUIRE_THROWS_WITH(std::make_shared<minifi::io::StreamSlice>(base, 7, 3), "StreamSlice is bigger than the Stream");
+ auto input_stream = std::static_pointer_cast<minifi::io::InputStream>(base);
+ REQUIRE_THROWS_WITH(std::make_shared<minifi::io::StreamSlice>(input_stream, 0, 9), "StreamSlice is bigger than the Stream");
+ REQUIRE_THROWS_WITH(std::make_shared<minifi::io::StreamSlice>(input_stream, 7, 3), "StreamSlice is bigger than the Stream");
}
TEST_CASE("StreamSliceTest1", "[teststreamslice]") {
std::shared_ptr<minifi::io::BaseStream> base = std::make_shared<minifi::io::BufferStream>();
base->write((const uint8_t*)"\x01\x02\x03\x04\x05\x06\x07\x08", 8);
- std::shared_ptr<minifi::io::BaseStream> stream_slice = std::make_shared<minifi::io::StreamSlice>(base, 2, 4);
+ auto input_stream = std::static_pointer_cast<minifi::io::InputStream>(base);
+ std::shared_ptr<minifi::io::InputStream> stream_slice = std::make_shared<minifi::io::StreamSlice>(input_stream, 2, 4);
std::vector<std::byte> buffer;
buffer.resize(stream_slice->size());
REQUIRE(stream_slice->read(buffer) == 4);
diff --git a/nanofi/include/cxx/CallbackProcessor.h b/nanofi/include/cxx/CallbackProcessor.h
index b5ae9c061..082e4c35f 100644
--- a/nanofi/include/cxx/CallbackProcessor.h
+++ b/nanofi/include/cxx/CallbackProcessor.h
@@ -30,7 +30,6 @@
#include <utility>
#include <sys/types.h>
#include "core/cstructs.h"
-#include "io/BaseStream.h"
#include "FlowFileRecord.h"
#include "core/Processor.h"
#include "core/ProcessSession.h"
diff --git a/nanofi/include/cxx/ReflexiveSession.h b/nanofi/include/cxx/ReflexiveSession.h
index 4804d9385..07232d3a1 100644
--- a/nanofi/include/cxx/ReflexiveSession.h
+++ b/nanofi/include/cxx/ReflexiveSession.h
@@ -15,8 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#ifndef __REFLEXIVE_SESSION_H__
-#define __REFLEXIVE_SESSION_H__
+#pragma once
#include <vector>
#include <queue>
@@ -28,47 +27,30 @@
#include "core/ProcessSession.h"
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
-namespace core {
+namespace org::apache::nifi::minifi::core {
-// ReflexiveSession Class
class ReflexiveSession : public ProcessSession{
public:
- // Constructor
- /*!
- * Create a new process session
- */
ReflexiveSession(std::shared_ptr<ProcessContext> processContext = nullptr)
- : ProcessSession(processContext) {
+ : ProcessSession(processContext) {
}
-// Destructor
- virtual ~ReflexiveSession() = default;
-
- virtual std::shared_ptr<core::FlowFile> get() {
- auto prevff = ff;
- ff = nullptr;
- return prevff;
- }
+ std::shared_ptr<core::FlowFile> get() override {
+ auto prevff = ff;
+ ff = nullptr;
+ return prevff;
+ }
- virtual void add(const std::shared_ptr<core::FlowFile> &flow) {
- ff = flow;
- }
- virtual void transfer(const std::shared_ptr<core::FlowFile>& /*flow*/, Relationship /*relationship*/) {
- // no op
- }
+ void add(const std::shared_ptr<core::FlowFile> &flow) override {
+ ff = flow;
+ }
+ void transfer(const std::shared_ptr<core::FlowFile>& /*flow*/, const Relationship& /*relationship*/) override {
+ // no op
+ }
protected:
- //
+
// Get the FlowFile from the highest priority queue
std::shared_ptr<core::FlowFile> ff;
};
-} /* namespace core */
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
-#endif
+} // namespace org::apache::nifi::minifi::core