Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
fgerlits closed pull request #2120: MINIFICPP-2731 Allow cancellation from ProcessSession::write URL: https://github.com/apache/nifi-minifi-cpp/pull/2120 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3008012372
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
I've opened a followup PR that addresses this issue.
https://github.com/apache/nifi-minifi-cpp/pull/2148
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3002880114
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
i've started the refactor but it snowballed and it seems way to big and
error prone to include it here without discarding existing reviews (im on 78
files modified and counting)
how about I do this refactor in a followup PR?
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
i've started the refactor but it snowballed and it seems way too big and
error prone to include it here without discarding existing reviews (im on 78
files modified and counting)
how about I do this refactor in a followup PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
fgerlits commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3001594370
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
yes, that sounds good
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3001526161
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
forcing the bytes written number into that enum feels weird, what if instead
std::expected? we can keep the normal int64_t on the C
api side and wrap it in an unexpected if its < 0 (we can even throw if the
return is out of the range of our defined (negative) error values)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
fgerlits commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r3000932337
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
Review Comment:
If someone is writing a new `io::OutputStreamCallback`, they will not know
that they can signal cancellation, or how to do it, unless they happen to find
and read this part of the source code.
Could we change `io::OutputStreamCallback` to return a `MinifiIoStatus`
instead of an integer, to make it easier for future developers?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853199097
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
+}
Review Comment:
[added unit
tests](https://github.com/apache/nifi-minifi-cpp/pull/2120/commits/f901307f1ec9950422eefc0a5c080c1da635fe82)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Copilot commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853135774
##
libminifi/src/core/ForwardingContentSession.cpp:
##
@@ -51,6 +51,11 @@ std::shared_ptr
ForwardingContentSession::append(const std::shar
return repository_->write(*resource_id, true);
}
+void ForwardingContentSession::remove(const std::shared_ptr&
resource_id) {
Review Comment:
The remove() method only erases the claim from the internal tracking sets
but doesn't delete the actual content from the repository. Since
ForwardingContentSession writes directly to the repository (line 43), when a
write is cancelled, the content file will remain on disk as an orphan. Add a
call to repository_->remove(*resource_id) to properly clean up the physical
file when remove() is called.
```suggestion
void ForwardingContentSession::remove(const std::shared_ptr&
resource_id) {
if (repository_ && resource_id) {
repository_->remove(*resource_id);
}
```
##
libminifi/test/libtest/unit/ContentRepositoryDependentTests.h:
##
@@ -171,4 +171,64 @@ void
testReadFromZeroLengthFlowFile(std::shared_ptr con
REQUIRE_NOTHROW(process_session.readBuffer(flow_file));
REQUIRE_NOTHROW(process_session.read(flow_file, ReadUntilItCan{}));
}
+
+inline void testErrWrite(std::shared_ptr
content_repo) {
+ const auto fixture = Fixture(std::move(content_repo));
+ core::ProcessSession& process_session = fixture.processSession();
+ const auto flow_file = process_session.create();
+ fixture.writeToFlowFile(flow_file, "original_content");
+ fixture.transferAndCommit(flow_file);
+
+ REQUIRE_THROWS(
+ process_session.write(flow_file, [](const
std::shared_ptr& output_stream) {
+output_stream->write("new_content");
+return MinifiIoStatus::MINIFI_IO_ERROR;
+ }));
+
+ CHECK(flow_file->getSize() == 16);
+ ReadUntilItCan read_until_it_can_callback;
+ const auto read_result = process_session.readBuffer(flow_file);
+ process_session.read(flow_file, std::ref(read_until_it_can_callback));
+ CHECK(to_string(read_result) == "original_content");
+}
+
+inline void testOkWrite(std::shared_ptr content_repo)
{
+ const auto fixture = Fixture(std::move(content_repo));
+ core::ProcessSession& process_session = fixture.processSession();
+ const auto flow_file = process_session.create();
+ fixture.writeToFlowFile(flow_file, "original_content");
+ fixture.transferAndCommit(flow_file);
+
+ CHECK(flow_file->getSize() == 16);
+
+ process_session.write(flow_file, [](const
std::shared_ptr& output_stream) {
+constexpr std::string str = "new_content";
Review Comment:
Using constexpr with std::string is unnecessary here. The string is not used
for compile-time computation and is only needed at runtime within the lambda.
Consider using a simple string literal, const char*, or std::string_view
instead of constexpr std::string for clarity and to avoid confusion about the
intended usage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
martinzink commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2853104528
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
Review Comment:
its C header so cant use enum class afaik
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] MINIFICPP-2731 Allow cancellation from ProcessSession::write [nifi-minifi-cpp]
Copilot commented on code in PR #2120:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2120#discussion_r2852764279
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
Review Comment:
`MinifiIoStatus` is declared as an unscoped enum in `minifi-c.h` (`typedef
enum MinifiIoStatus : int64_t { ... }`), so the enumerators are not members of
the type. `MinifiIoStatus::MINIFI_IO_CANCEL` will not compile; use
`MINIFI_IO_CANCEL` (unqualified) or change the enum to a scoped `enum class`
and adjust the comparison/casts accordingly.
##
libminifi/src/core/ProcessSession.cpp:
##
@@ -256,7 +257,15 @@ void ProcessSessionImpl::write(core::FlowFile &flow, const
io::OutputStreamCallb
if (nullptr == stream) {
throw Exception(FILE_OPERATION_EXCEPTION, "Failed to open flowfile
content for write");
}
-if (callback(stream) < 0) {
+const auto callback_result = callback(stream);
+if (callback_result == MinifiIoStatus::MINIFI_IO_CANCEL) {
+ stream->close();
+ content_session_->remove(claim);
+ claim.reset();
+ return;
+}
Review Comment:
The new cancellation path (`callback_result == ...CANCEL`) changes write
semantics but doesn't appear to be covered by existing unit/integration tests.
Please add a test that returns the cancel status from the callback and asserts
that the FlowFile's original content/claim is preserved and the temporary claim
is released (no orphan content left behind).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
