thyandrecardoso commented on code in PR #2018:
URL: https://github.com/apache/pekko-connectors/pull/2018#discussion_r4372405270
##########
amqp/src/test/scala/org/apache/pekko/stream/connectors/amqp/scaladsl/AmqpFlowSpec.scala:
##########
@@ -185,6 +186,52 @@ class AmqpFlowSpec extends AmqpSpec with AmqpMocking with
BeforeAndAfterEach {
messages should contain theSameElementsInOrderAs expectedResult
completion.futureValue shouldBe an[Done]
}
+
+ "preserve individual results when confirmations and rejections arrive out
of order" in assertAllStagesStopped {
+ val input = Vector("one", "two", "three", "four")
+ val (completion, probe, confirmCallback, rejectCallback) =
publishWithOrderedConfirm(input)
+
+ rejectCallback.handle(2L, false)
+ confirmCallback.handle(1L, false)
+ confirmCallback.handle(4L, false)
+ rejectCallback.handle(3L, false)
+
+ val expectedResult = Seq(
+ (WriteResult.confirmed, input(0)), (WriteResult.rejected, input(1)),
(WriteResult.rejected, input(2)),
+ (WriteResult.confirmed, input(3)))
+
+ probe.expectNextN(input.size) should contain theSameElementsInOrderAs
expectedResult
+ completion.futureValue shouldBe an[Done]
+ }
+
+ "preserve a rejection when a later batch confirmation includes the
message" in assertAllStagesStopped {
+ val input = Vector("one", "two", "three", "four")
+ val (completion, probe, confirmCallback, rejectCallback) =
publishWithOrderedConfirm(input)
+
+ rejectCallback.handle(2L, false)
+ // Message 4 is outside the batch but should be emitted once the earlier
messages have results.
+ confirmCallback.handle(4L, false)
+ confirmCallback.handle(3L, true)
+
+ val expectedResult = Seq(
+ (WriteResult.confirmed, input(0)), (WriteResult.rejected, input(1)),
(WriteResult.confirmed, input(2)),
+ (WriteResult.confirmed, input(3)))
+
+ probe.expectNextN(input.size) should contain theSameElementsInOrderAs
expectedResult
+ completion.futureValue shouldBe an[Done]
+ }
+
+ "preserve a confirmation when an earlier message times out" in
assertAllStagesStopped {
+ val input = Vector("one", "two")
+ val (completion, probe, confirmCallback, _) =
publishWithOrderedConfirm(input, confirmationTimeout = 200.millis)
Review Comment:
Sure. 55c4a69b3
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
