This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new bf818ed6e rewrite tests to avoid deprecated Source.queue methods
(#1856)
bf818ed6e is described below
commit bf818ed6e8f330465b0b39bd677442268698509c
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 1 11:57:51 2026 +0100
rewrite tests to avoid deprecated Source.queue methods (#1856)
---
.../jakartams/JmsProducerRetrySpec.scala | 24 +++++++++++++++++-----
.../connectors/jms/JmsProducerRetrySpec.scala | 24 +++++++++++++++++-----
2 files changed, 38 insertions(+), 10 deletions(-)
diff --git
a/jakartams/src/test/scala/org/apache/pekko/stream/connectors/jakartams/JmsProducerRetrySpec.scala
b/jakartams/src/test/scala/org/apache/pekko/stream/connectors/jakartams/JmsProducerRetrySpec.scala
index 4c84d94ed..8a606964f 100644
---
a/jakartams/src/test/scala/org/apache/pekko/stream/connectors/jakartams/JmsProducerRetrySpec.scala
+++
b/jakartams/src/test/scala/org/apache/pekko/stream/connectors/jakartams/JmsProducerRetrySpec.scala
@@ -16,9 +16,11 @@ package org.apache.pekko.stream.connectors.jakartams
import com.github.pjfanning.jakartamswrapper.WrappedConnectionFactory
import jakarta.jms.{ JMSException, Message, TextMessage }
import org.apache.pekko
+import pekko.Done
import pekko.stream._
import pekko.stream.connectors.jakartams.scaladsl.{ JmsConsumer, JmsProducer }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.testkit.TestProbe
import org.mockito.ArgumentMatchers.{ any, anyInt, anyLong }
import org.mockito.Mockito.when
import org.mockito.invocation.InvocationOnMock
@@ -50,8 +52,15 @@ class JmsProducerRetrySpec extends JmsSpec {
SendRetrySettings(system).withInitialRetry(10.millis).withMaxBackoff(10.millis).withInfiniteRetries()))
.withAttributes(ActorAttributes.supervisionStrategy(stoppingDecider))
- val (queue, result) = Source
- .queue[Int](10, OverflowStrategy.backpressure)
+ // one element is accepted at a time, each acknowledged back to `probe`
before the next is sent
+ val probe = TestProbe()
+ val ackMessage = "ack"
+
+ val (ref, result) = Source
+ .actorRefWithBackpressure[Int](
+ ackMessage,
+ { case Done => CompletionStrategy.draining },
+ PartialFunction.empty)
.zipWithIndex
.map(e => JmsMapMessage(Map("time" -> System.currentTimeMillis(),
"index" -> e._2)))
.via(jms)
@@ -64,7 +73,12 @@ class JmsProducerRetrySpec extends JmsSpec {
.take(20)
.runWith(Sink.seq)
- for (_ <- 1 to 10) queue.offer(1) // 10 before the crash
+ def offer(elem: Int): Unit = {
+ ref.tell(elem, probe.ref)
+ probe.expectMsg(20.seconds, ackMessage)
+ }
+
+ for (_ <- 1 to 10) offer(1) // 10 before the crash
Thread.sleep(500)
server.stop() // crash.
@@ -72,8 +86,8 @@ class JmsProducerRetrySpec extends JmsSpec {
// https://activemq.apache.org/how-do-i-restart-embedded-broker.html
server.start() // recover.
val restartTime = System.currentTimeMillis()
- for (_ <- 1 to 10) queue.offer(1) // 10 after the crash
- queue.complete()
+ for (_ <- 1 to 10) offer(1) // 10 after the crash
+ ref.tell(Done, probe.ref)
val resultList = result.futureValue
def index(m: Map[String, Any]) = m("index").asInstanceOf[Long]
diff --git
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
index 1aa0fb1b8..dd8ce13c1 100644
---
a/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
+++
b/jms/src/test/scala/org/apache/pekko/stream/connectors/jms/JmsProducerRetrySpec.scala
@@ -16,9 +16,11 @@ package org.apache.pekko.stream.connectors.jms
import java.util.concurrent.atomic.AtomicInteger
import org.apache.pekko
+import pekko.Done
import pekko.stream._
import pekko.stream.connectors.jms.scaladsl.{ JmsConsumer, JmsProducer }
import pekko.stream.scaladsl.{ Keep, Sink, Source }
+import pekko.testkit.TestProbe
import com.github.pjfanning.jmswrapper.WrappedConnectionFactory
import javax.jms.{ JMSException, Message, TextMessage }
@@ -52,8 +54,15 @@ class JmsProducerRetrySpec extends JmsSpec {
SendRetrySettings(system).withInitialRetry(10.millis).withMaxBackoff(10.millis).withInfiniteRetries()))
.withAttributes(ActorAttributes.supervisionStrategy(stoppingDecider))
- val (queue, result) = Source
- .queue[Int](10, OverflowStrategy.backpressure)
+ // one element is accepted at a time, each acknowledged back to `probe`
before the next is sent
+ val probe = TestProbe()
+ val ackMessage = "ack"
+
+ val (ref, result) = Source
+ .actorRefWithBackpressure[Int](
+ ackMessage,
+ { case Done => CompletionStrategy.draining },
+ PartialFunction.empty)
.zipWithIndex
.map(e => JmsMapMessage(Map("time" -> System.currentTimeMillis(),
"index" -> e._2)))
.via(jms)
@@ -66,7 +75,12 @@ class JmsProducerRetrySpec extends JmsSpec {
.take(20)
.runWith(Sink.seq)
- for (_ <- 1 to 10) queue.offer(1) // 10 before the crash
+ def offer(elem: Int): Unit = {
+ ref.tell(elem, probe.ref)
+ probe.expectMsg(20.seconds, ackMessage)
+ }
+
+ for (_ <- 1 to 10) offer(1) // 10 before the crash
Thread.sleep(500)
server.stop() // crash.
@@ -75,8 +89,8 @@ class JmsProducerRetrySpec extends JmsSpec {
server.service.waitUntilStopped()
server.start(true) // recover.
val restartTime = System.currentTimeMillis()
- for (_ <- 1 to 10) queue.offer(1) // 10 after the crash
- queue.complete()
+ for (_ <- 1 to 10) offer(1) // 10 after the crash
+ ref.tell(Done, probe.ref)
val resultList = result.futureValue
def index(m: Map[String, Any]) = m("index").asInstanceOf[Long]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
