Merge pull request #257 from fp-in-bo/more-stream-concurrency

Improve stream parjoin
fp-in-bo · Dec 21, 2021 · fd324b2 · fd324b2
2 parents a4d1ad9 + 7964163
commit fd324b2
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 17 deletions.
diff --git a/core/src/main/scala/jms4s/JmsAcknowledgerConsumer.scala b/core/src/main/scala/jms4s/JmsAcknowledgerConsumer.scala
@@ -65,14 +65,9 @@ object JmsAcknowledgerConsumer {
   ): JmsAcknowledgerConsumer[F] =
     (f: (JmsMessage, MessageFactory[F]) => F[AckAction[F]]) => {
       Stream
-        .emits(0 until concurrencyLevel)
-        .as(
-          Stream.eval(
-            pool.receive(f)
-          )
-        )
-        .parJoin(concurrencyLevel)
+        .emit(Stream.eval(pool.receive(f)))
         .repeat
+        .parJoin(concurrencyLevel)
         .compile
         .drain
     }

diff --git a/core/src/main/scala/jms4s/JmsAutoAcknowledgerConsumer.scala b/core/src/main/scala/jms4s/JmsAutoAcknowledgerConsumer.scala
@@ -66,12 +66,9 @@ object JmsAutoAcknowledgerConsumer {
   ): JmsAutoAcknowledgerConsumer[F] =
     (f: (JmsMessage, MessageFactory[F]) => F[AutoAckAction[F]]) =>
       Stream
-        .emits(0 until concurrencyLevel)
-        .as(
-          Stream.eval(pool.receive(f))
-        )
-        .parJoin(concurrencyLevel)
+        .emit(Stream.eval(pool.receive(f)))
         .repeat
+        .parJoin(concurrencyLevel)
         .compile
         .drain
 

diff --git a/core/src/main/scala/jms4s/JmsTransactedConsumer.scala b/core/src/main/scala/jms4s/JmsTransactedConsumer.scala
@@ -66,12 +66,9 @@ object JmsTransactedConsumer {
   ): JmsTransactedConsumer[F] =
     (f: (JmsMessage, MessageFactory[F]) => F[TransactionAction[F]]) =>
       Stream
-        .emits(0 until concurrencyLevel)
-        .as(
-          Stream.eval(pool.receive(f))
-        )
-        .parJoin(concurrencyLevel)
+        .emit(Stream.eval(pool.receive(f)))
         .repeat
+        .parJoin(concurrencyLevel)
         .compile
         .drain