You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2025/07/07 03:06:51 UTC
(pekko) branch main updated: feat: Add Source#create method (#1823)
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 60a84ce2e2 feat: Add Source#create method (#1823)
60a84ce2e2 is described below
commit 60a84ce2e2b02d227b78deef688ad51e17a0a038
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Mon Jul 7 11:06:46 2025 +0800
feat: Add Source#create method (#1823)
---
.../apache/pekko/stream/javadsl/SourceTest.java | 16 ++++++++
.../stream/scaladsl/BoundedSourceQueueSpec.scala | 14 +++++++
.../org/apache/pekko/stream/impl/Stages.scala | 1 +
.../org/apache/pekko/stream/javadsl/Source.scala | 34 +++++++++++++++++
.../org/apache/pekko/stream/scaladsl/Source.scala | 44 ++++++++++++++++++++++
5 files changed, 109 insertions(+)
diff --git a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
index f3ffc0fd45..e6bbd01be0 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceTest.java
@@ -1620,4 +1620,20 @@ public class SourceTest extends StreamTest {
Pair.create(4, 1L))),
resultSet);
}
+
+ @Test
+ public void createSourceWithCreateMethod() throws Exception {
+ Source.<String>create(
+ 100,
+ queue -> {
+ queue.offer("Message1");
+ queue.offer("Message2");
+ queue.complete();
+ })
+ .runWith(TestSink.probe(system), system)
+ .ensureSubscription()
+ .request(2)
+ .expectNext("Message1", "Message2")
+ .expectComplete();
+ }
}
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/BoundedSourceQueueSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/BoundedSourceQueueSpec.scala
index 738e84d965..714b472ab6 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/BoundedSourceQueueSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/BoundedSourceQueueSpec.scala
@@ -223,5 +223,19 @@ class BoundedSourceQueueSpec extends StreamSpec("""pekko.loglevel = debug
}
queue.size() shouldBe 100
}
+
+ "can be used to create create method" in {
+ Source.create[String](1024)(queue => {
+ queue.offer("a")
+ queue.offer("b")
+ queue.complete()
+ })
+ .toMat(TestSink.probe[String])(Keep.right)
+ .run()
+ .ensureSubscription()
+ .request(2)
+ .expectNext("a", "b")
+ .expectComplete()
+ }
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
index 1d2c872a6e..00278f7a41 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Stages.scala
@@ -135,6 +135,7 @@ import pekko.stream.Attributes._
val actorRefSource = name("actorRefSource")
val actorRefWithBackpressureSource = name("actorRefWithBackpressureSource")
val queueSource = name("queueSource")
+ val create = name("create")
val inputStreamSource = name("inputStreamSource") and IODispatcher
val outputStreamSource = name("outputStreamSource")
val fileSource = name("fileSource") and IODispatcher
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 7be481b4c5..fe3e6c694b 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -798,6 +798,40 @@ object Source {
def queue[T](bufferSize: Int): Source[T, BoundedSourceQueue[T]] =
scaladsl.Source.queue(bufferSize).asJava
+ /**
+ * Creates a Source that will immediately execute the provided function `producer` with a [[BoundedSourceQueue]] when materialized.
+ * This allows defining element production logic at Source creation time.
+ *
+ * The function `producer` can push elements to the stream using the provided queue. The queue behaves the same as in [[Source.queue]]:
+ * <br>
+ * - Elements are emitted when there is downstream demand, buffered otherwise
+ * <br>
+ * - Elements are dropped if the buffer is full
+ * <br>
+ * - Buffered elements are discarded if downstream terminates
+ * <br>
+ * You should never block the producer thread, as it will block the stream from processing elements.
+ * If the function `producer` throws an exception, the queue will be failed and the exception will be propagated to the stream.
+ *
+ * Example usage:
+ * {{{
+ * Source.create[Int](10) { queue =>
+ * // This code is executed when the source is materialized
+ * queue.offer(1)
+ * queue.offer(2)
+ * queue.offer(3)
+ * queue.complete()
+ * }
+ * }}}
+ *
+ * @param bufferSize the size of the buffer (number of elements)
+ * @param producer function that receives the queue and defines how to produce data
+ * @return a Source that emits elements pushed to the queue
+ * @since 1.2.0
+ */
+ def create[T](bufferSize: Int, producer: function.Procedure[BoundedSourceQueue[T]]): Source[T, NotUsed] =
+ scaladsl.Source.create(bufferSize)(producer(_)).asJava
+
/**
* Creates a `Source` that is materialized as an [[pekko.stream.javadsl.SourceQueueWithComplete]].
* You can push elements to the queue and they will be emitted to the stream if there is demand from downstream,
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
index 33825f2ddf..5e67380fd8 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala
@@ -20,6 +20,7 @@ import scala.annotation.unchecked.uncheckedVariance
import scala.collection.{ immutable, AbstractIterator }
import scala.concurrent.{ Future, Promise }
import scala.concurrent.duration.FiniteDuration
+import scala.util.control.NonFatal
import org.apache.pekko
import pekko.{ Done, NotUsed }
@@ -930,6 +931,49 @@ object Source {
def queue[T](bufferSize: Int): Source[T, BoundedSourceQueue[T]] =
Source.fromGraph(new BoundedSourceQueueStage[T](bufferSize))
+ /**
+ * Creates a Source that will immediately execute the provided function `producer` with a [[BoundedSourceQueue]] when materialized.
+ * This allows defining element production logic at Source creation time.
+ *
+ * The function `producer` can push elements to the stream using the provided queue. The queue behaves the same as in [[Source.queue]]:
+ * <br>
+ * - Elements are emitted when there is downstream demand, buffered otherwise
+ * <br>
+ * - Elements are dropped if the buffer is full
+ * <br>
+ * - Buffered elements are discarded if downstream terminates
+ * <br>
+ * You should never block the producer thread, as it will block the stream from processing elements.
+ * If the function `producer` throws an exception, the queue will be failed and the exception will be propagated to the stream.
+ *
+ * Example usage:
+ * {{{
+ * Source.create[Int](10) { queue =>
+ * // This code is executed when the source is materialized
+ * queue.offer(1)
+ * queue.offer(2)
+ * queue.offer(3)
+ * queue.complete()
+ * }
+ * }}}
+ *
+ * @param bufferSize the size of the buffer (number of elements)
+ * @param producer function that receives the queue and defines how to produce data
+ * @return a Source that emits elements pushed to the queue
+ * @since 1.2.0
+ */
+ def create[T](bufferSize: Int)(producer: BoundedSourceQueue[T] => Unit): Source[T, NotUsed] =
+ Source.fromGraph(new BoundedSourceQueueStage[T](bufferSize).mapMaterializedValue(queue => {
+ try {
+ producer(queue)
+ NotUsed
+ } catch {
+ case NonFatal(e) =>
+ queue.fail(e)
+ NotUsed
+ }
+ }).withAttributes(DefaultAttributes.create))
+
/**
* Creates a `Source` that is materialized as an [[pekko.stream.scaladsl.SourceQueueWithComplete]].
* You can push elements to the queue and they will be emitted to the stream if there is demand from downstream,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org