You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by en...@apache.org on 2024/02/09 08:25:03 UTC
(incubator-pekko-connectors) branch main updated: mqtt-streaming: fix flaky 'duplicate publish given two client connections' test (#456)
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 48e7b34cf mqtt-streaming: fix flaky 'duplicate publish given two client connections' test (#456)
48e7b34cf is described below
commit 48e7b34cf8ad021a0771ecd6c58aa28a4d0117f7
Author: Arnout Engelen <ar...@bzzt.net>
AuthorDate: Fri Feb 9 09:24:58 2024 +0100
mqtt-streaming: fix flaky 'duplicate publish given two client connections' test (#456)
It appears the flakiness was caused by the two connections using the same
connection id for different connections in the same session. I'm not intimately
familiar with this codebase, but AFAICS it seems this connection id is an
internal concern to the component, and not an MQTT concept. I think we should
probably provide an API that does not bother the user with this concept at
all, and manages non-overlapping connection ids entirely internal to the
component. I don't get the impression that it was the purpose of the test to
test the behaviour in case of overlapping connection ids.
Fixes #454
After this change the 'unsubscribe a client session' seems to fail
a lot on CI, but that seems unrelated and something to fix in a separate PR.
---
.../test/scala/docs/scaladsl/MqttSessionSpec.scala | 247 ++++++++++-----------
1 file changed, 122 insertions(+), 125 deletions(-)
diff --git a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
index 237f301fa..38fd32dc0 100644
--- a/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
+++ b/mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala
@@ -32,7 +32,7 @@ import pekko.testkit._
import pekko.util.{ ByteString, Timeout }
import org.scalatest.BeforeAndAfterAll
import org.scalatest.concurrent.{ Eventually, ScalaFutures }
-import org.scalatest.time.{ Millis, Minutes, Span }
+import org.scalatest.time.{ Millis, Span }
import scala.concurrent.{ ExecutionContext, Promise }
import scala.concurrent.duration._
@@ -1922,159 +1922,156 @@ class MqttSessionSpec
// longer patience needed since Akka 2.6
implicit val patienceConfig: PatienceConfig = PatienceConfig(scaled(1.second), scaled(50.millis))
- // https://github.com/apache/incubator-pekko-connectors/issues/148
- eventually(timeout(Span(1, Minutes))) {
- val serverSession = ActorMqttServerSession(settings.withProducerPubAckRecTimeout(10.millis))
+ val serverSession = ActorMqttServerSession(settings.withProducerPubAckRecTimeout(10.millis))
- val client1 = TestProbe()
- val toClient1 = Sink.foreach[ByteString](bytes => client1.ref ! bytes)
- val (client1Connection, fromClient1) = Source
- .queue[ByteString](1, OverflowStrategy.dropHead)
- .toMat(BroadcastHub.sink)(Keep.both)
- .run()
+ val client1 = TestProbe()
+ val toClient1 = Sink.foreach[ByteString](bytes => client1.ref ! bytes)
+ val (client1Connection, fromClient1) = Source
+ .queue[ByteString](1, OverflowStrategy.dropHead)
+ .toMat(BroadcastHub.sink)(Keep.both)
+ .run()
- val pipeToClient1 = Flow.fromSinkAndSource(toClient1, fromClient1)
+ val pipeToClient1 = Flow.fromSinkAndSource(toClient1, fromClient1)
- val client2 = TestProbe()
- val toClient2 = Sink.foreach[ByteString](bytes => client2.ref ! bytes)
- val (client2Connection, fromClient2) = Source
- .queue[ByteString](0, OverflowStrategy.dropHead)
- .toMat(BroadcastHub.sink)(Keep.both)
- .run()
+ val client2 = TestProbe()
+ val toClient2 = Sink.foreach[ByteString](bytes => client2.ref ! bytes)
+ val (client2Connection, fromClient2) = Source
+ .queue[ByteString](0, OverflowStrategy.dropHead)
+ .toMat(BroadcastHub.sink)(Keep.both)
+ .run()
- val pipeToClient2 = Flow.fromSinkAndSource(toClient2, fromClient2)
+ val pipeToClient2 = Flow.fromSinkAndSource(toClient2, fromClient2)
- val clientId = "some-client-id"
+ val clientId = "some-client-id"
+
+ val connect = Connect(clientId, ConnectFlags.None)
+ val connect1Received = Promise[Done]()
+ val connect2Received = Promise[Done]()
- val connect = Connect(clientId, ConnectFlags.None)
- val connect1Received = Promise[Done]()
- val connect2Received = Promise[Done]()
+ val subscribe = Subscribe("some-topic")
+ val subscribe1Received = Promise[Done]()
+ val subscribe2Received = Promise[Done]()
- val subscribe = Subscribe("some-topic")
- val subscribe1Received = Promise[Done]()
- val subscribe2Received = Promise[Done]()
+ val pubAckReceived = Promise[Done]()
- val pubAckReceived = Promise[Done]()
+ val disconnect = Disconnect
+ val disconnectReceived = Promise[Done]()
- val disconnect = Disconnect
- val disconnectReceived = Promise[Done]()
+ val serverConnection1 =
+ Source
+ .queue[Command[Nothing]](1, OverflowStrategy.fail)
+ .via(
+ Mqtt
+ .serverSessionFlow(serverSession, ByteString("connection 1"))
+ .join(pipeToClient1))
+ .wireTap(Sink.foreach[Either[DecodeError, Event[_]]] {
+ case Right(Event(`connect`, _)) =>
+ connect1Received.success(Done)
+ case Right(Event(cp: Subscribe, _)) if cp.topicFilters == subscribe.topicFilters =>
+ subscribe1Received.success(Done)
+ case Right(Event(`disconnect`, _)) =>
+ disconnectReceived.success(Done)
+ case other => fail(s"didn't match `$other`")
+ })
+ .toMat(Sink.seq)(Keep.left)
+ .run()
+
+ val connectBytes = connect.encode(ByteString.newBuilder).result()
+ val connAck = ConnAck(ConnAckFlags.None, ConnAckReturnCode.ConnectionAccepted)
+ val connAckBytes = connAck.encode(ByteString.newBuilder).result()
+
+ val subscribeBytes = subscribe.encode(ByteString.newBuilder, PacketId(1)).result()
+ val subAck = SubAck(PacketId(1), List(ControlPacketFlags.QoSAtLeastOnceDelivery))
+ val subAckBytes = subAck.encode(ByteString.newBuilder).result()
- val serverConnection1 =
- Source
- .queue[Command[Nothing]](1, OverflowStrategy.fail)
- .via(
- Mqtt
- .serverSessionFlow(serverSession, ByteString.empty)
- .join(pipeToClient1))
- .wireTap(Sink.foreach[Either[DecodeError, Event[_]]] {
- case Right(Event(`connect`, _)) =>
- connect1Received.success(Done)
- case Right(Event(cp: Subscribe, _)) if cp.topicFilters == subscribe.topicFilters =>
- subscribe1Received.success(Done)
- case Right(Event(`disconnect`, _)) =>
- disconnectReceived.success(Done)
- case other => fail(s"didn't match `$other`")
- })
- .toMat(Sink.seq)(Keep.left)
- .run()
+ val publish = Publish("some-topic", ByteString("some-payload"))
+ val publishBytes = publish.encode(ByteString.newBuilder, Some(PacketId(1))).result()
+ val dupPublishBytes = publish
+ .copy(flags = publish.flags | ControlPacketFlags.DUP)
+ .encode(ByteString.newBuilder, Some(PacketId(1)))
+ .result()
+ val pubAck = PubAck(PacketId(1))
+ val pubAckBytes = pubAck.encode(ByteString.newBuilder).result()
- val connectBytes = connect.encode(ByteString.newBuilder).result()
- val connAck = ConnAck(ConnAckFlags.None, ConnAckReturnCode.ConnectionAccepted)
- val connAckBytes = connAck.encode(ByteString.newBuilder).result()
+ val disconnectBytes = disconnect.encode(ByteString.newBuilder).result()
- val subscribeBytes = subscribe.encode(ByteString.newBuilder, PacketId(1)).result()
- val subAck = SubAck(PacketId(1), List(ControlPacketFlags.QoSAtLeastOnceDelivery))
- val subAckBytes = subAck.encode(ByteString.newBuilder).result()
+ client1Connection.offer(connectBytes)
- val publish = Publish("some-topic", ByteString("some-payload"))
- val publishBytes = publish.encode(ByteString.newBuilder, Some(PacketId(1))).result()
- val dupPublishBytes = publish
- .copy(flags = publish.flags | ControlPacketFlags.DUP)
- .encode(ByteString.newBuilder, Some(PacketId(1)))
- .result()
- val pubAck = PubAck(PacketId(1))
- val pubAckBytes = pubAck.encode(ByteString.newBuilder).result()
+ connect1Received.future.futureValue shouldBe Done
- val disconnectBytes = disconnect.encode(ByteString.newBuilder).result()
+ serverConnection1.offer(Command(connAck))
+ client1.expectMsg(connAckBytes)
- client1Connection.offer(connectBytes)
+ client1Connection.offer(subscribeBytes)
- connect1Received.future.futureValue shouldBe Done
+ subscribe1Received.future.futureValue shouldBe Done
- serverConnection1.offer(Command(connAck))
- client1.expectMsg(connAckBytes)
+ serverConnection1.offer(Command(subAck))
+ client1.expectMsg(subAckBytes)
- client1Connection.offer(subscribeBytes)
+ serverSession ! Command(publish)
+ client1.expectMsg(publishBytes)
- subscribe1Received.future.futureValue shouldBe Done
+ // Perform an explicit disconnect otherwise, if for example, we
+ // just completed the client connection, the session may receive
+ // the associated ConnectionLost signal for the new connection
+ // given that the new connection occurs so quickly.
+ client1Connection.offer(disconnectBytes)
- serverConnection1.offer(Command(subAck))
- client1.expectMsg(subAckBytes)
+ disconnectReceived.future.futureValue shouldBe Done
- serverSession ! Command(publish)
- client1.expectMsg(publishBytes)
+ val serverConnection2 =
+ Source
+ .queue[Command[Nothing]](1, OverflowStrategy.fail)
+ .via(
+ Mqtt
+ .serverSessionFlow(serverSession, ByteString("connection 2"))
+ .join(pipeToClient2))
+ .wireTap(Sink.foreach[Either[DecodeError, Event[_]]] {
+ case Right(Event(`connect`, _)) =>
+ connect2Received.success(Done)
+ case Right(Event(cp: Subscribe, _)) if cp.topicFilters == subscribe.topicFilters =>
+ subscribe2Received.success(Done)
+ case Right(Event(_: PubAck, _)) =>
+ pubAckReceived.success(Done)
+ case other => fail(s"didn't match `$other`")
+ })
+ .toMat(Sink.seq)(Keep.left)
+ .run()
- // Perform an explicit disconnect otherwise, if for example, we
- // just completed the client connection, the session may receive
- // the associated ConnectionLost signal for the new connection
- // given that the new connection occurs so quickly.
- client1Connection.offer(disconnectBytes)
+ client2Connection.offer(connectBytes)
- disconnectReceived.future.futureValue shouldBe Done
+ connect2Received.future.futureValue shouldBe Done
- val serverConnection2 =
- Source
- .queue[Command[Nothing]](1, OverflowStrategy.fail)
- .via(
- Mqtt
- .serverSessionFlow(serverSession, ByteString.empty)
- .join(pipeToClient2))
- .wireTap(Sink.foreach[Either[DecodeError, Event[_]]] {
- case Right(Event(`connect`, _)) =>
- connect2Received.success(Done)
- case Right(Event(cp: Subscribe, _)) if cp.topicFilters == subscribe.topicFilters =>
- subscribe2Received.success(Done)
- case Right(Event(_: PubAck, _)) =>
- pubAckReceived.success(Done)
- case other => fail(s"didn't match `$other`")
- })
- .toMat(Sink.seq)(Keep.left)
- .run()
-
- client2Connection.offer(connectBytes)
-
- connect2Received.future.futureValue shouldBe Done
-
- serverConnection2.offer(Command(connAck))
- client2.expectMsg(6.seconds, connAckBytes)
-
- client2Connection.offer(subscribeBytes)
-
- subscribe2Received.future.futureValue shouldBe Done
-
- serverConnection2.offer(Command(subAck))
-
- client2.fishForMessage(3.seconds.dilated) {
- case msg: ByteString if msg == dupPublishBytes => true
- case _ => false
- }
+ serverConnection2.offer(Command(connAck))
+ client2.expectMsg(6.seconds, connAckBytes)
- client2Connection.offer(pubAckBytes)
- pubAckReceived.future.futureValue shouldBe Done
+ client2Connection.offer(subscribeBytes)
- client1Connection.complete()
- client2Connection.complete()
- serverConnection1.complete()
- serverConnection2.complete()
+ subscribe2Received.future.futureValue shouldBe Done
- for {
- _ <- client1Connection.watchCompletion()
- _ <- client2Connection.watchCompletion()
- _ <- serverConnection1.watchCompletion()
- _ <- serverConnection2.watchCompletion()
- } serverSession.shutdown()
+ serverConnection2.offer(Command(subAck))
+ client2.fishForMessage(3.seconds.dilated) {
+ case msg: ByteString if msg == dupPublishBytes => true
+ case _ => false
}
+
+ client2Connection.offer(pubAckBytes)
+ pubAckReceived.future.futureValue shouldBe Done
+
+ client1Connection.complete()
+ client2Connection.complete()
+ serverConnection1.complete()
+ serverConnection2.complete()
+
+ for {
+ _ <- client1Connection.watchCompletion()
+ _ <- client2Connection.watchCompletion()
+ _ <- serverConnection1.watchCompletion()
+ _ <- serverConnection2.watchCompletion()
+ } serverSession.shutdown()
+
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org