(incubator-pekko-connectors) branch main updated: mqtt-streaming: fix flaky 'unsubscribe a client session' test (#459)

[en...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

engelen pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko-connectors.git


The following commit(s) were added to refs/heads/main by this push:
     new 7c05e1067 mqtt-streaming: fix flaky 'unsubscribe a client session' test (#459)
7c05e1067 is described below

commit 7c05e1067aa17f1a370bc5d918af9f7977fc4449
Author: Arnout Engelen <ar...@bzzt.net>
AuthorDate: Fri Feb 9 09:25:12 2024 +0100

    mqtt-streaming: fix flaky 'unsubscribe a client session' test (#459)
    
    Previously, in this test, the server sent a 'publish'
    concurrent to the client sending the 'ack' for that
    publish. This fix makes sure the 'ack' is sent only after
    observing the 'publish'.
    
    Hopefully fixes #458
---
 .../src/test/scala/docs/scaladsl/MqttSessionSpec.scala    | 15 ++++++---------
 1 file changed, 6 insertions(+), 9 deletions(-)

diff --git a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
index 38fd32dc0..385ab54e3 100644
--- a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
+++ b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
@@ -1173,7 +1173,7 @@ class MqttSessionSpec
       val unsubAck = UnsubAck(PacketId(1))
       val unsubAckReceived = Promise[Done]()
 
-      val (client, result) =
+      val (client, results) =
         Source
           .queue(2, OverflowStrategy.fail)
           .via(
@@ -1188,11 +1188,7 @@ class MqttSessionSpec
               case _ =>
             }
           }
-          .takeWhile {
-            case Right(Event(PingResp, None)) => false
-            case _                            => true
-          }
-          .toMat(Sink.seq)(Keep.both)
+          .toMat(Sink.queue(10))(Keep.both)
           .run()
 
       val connect = Connect("some-client-id", ConnectFlags.None)
@@ -1232,18 +1228,19 @@ class MqttSessionSpec
       client.offer(Command(unsubscribe))
 
       server.expectMsg(unsubscribeBytes)
+      // Possible to receive a pub from an unsubscribed topic given that it may be in transit
       server.reply(unsubAckBytes ++ publishDupBytes)
 
       unsubAckReceived.future.futureValue shouldBe Done
 
+      results.queue.pull().futureValue shouldBe Some(Right(Event(unsubAck)))
+      results.queue.pull().futureValue shouldBe Some(Right(Event(publishDup)))
       client.offer(Command(pubAck))
 
       server.expectMsg(pubAckBytes)
       server.reply(pingRespBytes)
 
-      // Quite possible to receive a pub from an unsubscribed topic given that it may be in transit
-      result.futureValue shouldBe Vector(Right(Event(unsubAck)), Right(Event(publishDup)))
-
+      results.cancel()
       client.complete()
       client.watchCompletion().foreach(_ => session.shutdown())
     }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org