-
Notifications
You must be signed in to change notification settings - Fork 3.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Convenience stream operators for publishing and subscribing to typed …
…pub sub #31037
- Loading branch information
1 parent
ccd5219
commit b9a1a66
Showing
8 changed files
with
283 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
37 changes: 37 additions & 0 deletions
37
akka-docs/src/main/paradox/stream/operators/PubSub/sink.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,37 @@ | ||
# PubSub.sink | ||
|
||
A sink that will publish emitted messages to a @apidoc[akka.actor.typed.pubsub.Topic$]. | ||
|
||
@ref[Actor interop operators](../index.md#actor-interop-operators) | ||
|
||
Note that there is no backpressure from the topic, so care must be taken to not publish messages at a higher rate than that can be handled | ||
by subscribers. | ||
|
||
If the topic does not have any subscribers when a message is published, or the topic actor is stopped, the message is sent to dead letters. | ||
|
||
## Dependency | ||
|
||
This operator is included in: | ||
|
||
@@dependency[sbt,Maven,Gradle] { | ||
bomGroup=com.typesafe.akka bomArtifact=akka-bom_$scala.binary.version$ bomVersionSymbols=AkkaVersion | ||
symbol1=AkkaVersion | ||
value1="$akka.version$" | ||
group="com.typesafe.akka" | ||
artifact="akka-stream-typed_$scala.binary.version$" | ||
version=AkkaVersion | ||
} | ||
|
||
## Signature | ||
|
||
@apidoc[PubSub.sink](akka.stream.typed.*.PubSub$) { scala="#sink[T](topic:akka.actor.typed.Toppic[T]):akka.stream.scaladsl.Sink[T,akka.NotUsed]" java="#sink(akka.actor.typed.Topic)" } | ||
|
||
## Reactive Streams semantics | ||
|
||
@@@div { .callout } | ||
|
||
**cancels** never | ||
|
||
**backpressures** never | ||
|
||
@@@ |
40 changes: 40 additions & 0 deletions
40
akka-docs/src/main/paradox/stream/operators/PubSub/source.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,40 @@ | ||
# PubSub.source | ||
|
||
A source that will subscribe to a @apidoc[akka.actor.typed.pubsub.Topic$] and stream messages published to the topic. | ||
|
||
@ref[Actor interop operators](../index.md#actor-interop-operators) | ||
|
||
The source can be materialized multiple times, each materialized stream will stream messages published to the topic after the stream has started. | ||
|
||
Note that it is not possible to propagate the backpressure from the running stream to the pub sub topic, | ||
if the stream is backpressuring published messages are buffered up to a limit and if the limit is hit | ||
the configurable `OverflowStrategy` decides what happens. It is not possible to use the `Backpressure` | ||
strategy. | ||
|
||
|
||
## Dependency | ||
|
||
This operator is included in: | ||
|
||
@@dependency[sbt,Maven,Gradle] { | ||
bomGroup=com.typesafe.akka bomArtifact=akka-bom_$scala.binary.version$ bomVersionSymbols=AkkaVersion | ||
symbol1=AkkaVersion | ||
value1="$akka.version$" | ||
group="com.typesafe.akka" | ||
artifact="akka-stream-typed_$scala.binary.version$" | ||
version=AkkaVersion | ||
} | ||
|
||
## Signature | ||
|
||
@apidoc[PubSub.source](akka.stream.typed.*.PubSub$) { scala="#source[T](topic:akka.actor.typed.Toppic[T]):akka.stream.scaladsl.Source[T,akka.NotUsed]" java="#source(akka.actor.typed.Topic)" } | ||
|
||
## Reactive Streams semantics | ||
|
||
@@@div { .callout } | ||
|
||
**emits** a message published to the topic is emitted as soon as there is demand from downstream | ||
|
||
**completes** when the topic actor terminates | ||
|
||
@@@ |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
55 changes: 55 additions & 0 deletions
55
akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/PubSub.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,55 @@ | ||
/* | ||
* Copyright (C) 2009-2021 Lightbend Inc. <https://www.lightbend.com> | ||
*/ | ||
|
||
package akka.stream.typed.javadsl | ||
|
||
import akka.NotUsed | ||
import akka.actor.typed.ActorRef | ||
import akka.actor.typed.pubsub.Topic | ||
import akka.annotation.ApiMayChange | ||
import akka.stream.OverflowStrategy | ||
import akka.stream.javadsl.Sink | ||
import akka.stream.javadsl.Source | ||
|
||
/** | ||
* Sources and sinks to integrate [[akka.actor.typed.pubsub.Topic]] with streams allowing for local or distributed | ||
* publishing and subscribing of elements through a stream. | ||
*/ | ||
object PubSub { | ||
|
||
/** | ||
* Create a source that will subscribe to a topic and stream messages published to the topic. Can be materialized | ||
* multiple times, each materialized stream will contain messages published after it was started. | ||
* | ||
* Note that it is not possible to propagate the backpressure from the running stream to the pub sub topic, | ||
* if the stream is backpressuring published messages are buffered up to a limit and if the limit is hit | ||
* the configurable `OverflowStrategy` decides what happens. It is not possible to use the `Backpressure` | ||
* strategy. | ||
* | ||
* @param topicActor The actor ref for an `akka.actor.typed.pubsub.Topic` actor representing a specific topic. | ||
* @param bufferSize The maximum number of messages to buffer if the stream applies backpressure | ||
* @param overflowStrategy Strategy to use once the buffer is full. | ||
* @tparam T The type of the published messages | ||
*/ | ||
@ApiMayChange | ||
def source[T]( | ||
topicActor: ActorRef[Topic.Command[T]], | ||
bufferSize: Int, | ||
overflowStrategy: OverflowStrategy): Source[T, NotUsed] = | ||
akka.stream.typed.scaladsl.PubSub.source(topicActor, bufferSize, overflowStrategy).asJava | ||
|
||
/** | ||
* Create a sink that will publish each message to the given topic. Note that there is no backpressure | ||
* from the topic, so care must be taken to not publish messages at a higher rate than that can be handled | ||
* by subscribers. If the topic does not have any subscribers when a message is published the message is | ||
* sent to dead letters. | ||
* | ||
* @param topicActor The actor ref for an `akka.actor.typed.pubsub.Topic` actor representing a specific topic. | ||
* @tparam T the type of the messages that can be published | ||
*/ | ||
@ApiMayChange | ||
def sink[T](topicActor: ActorRef[Topic.Command[T]]): Sink[T, NotUsed] = | ||
akka.stream.typed.scaladsl.PubSub.sink[T](topicActor).asJava | ||
|
||
} |
64 changes: 64 additions & 0 deletions
64
akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/PubSub.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
/* | ||
* Copyright (C) 2009-2021 Lightbend Inc. <https://www.lightbend.com> | ||
*/ | ||
|
||
package akka.stream.typed.scaladsl | ||
|
||
import akka.NotUsed | ||
import akka.actor.typed.ActorRef | ||
import akka.actor.typed.pubsub.Topic | ||
import akka.annotation.ApiMayChange | ||
import akka.stream.OverflowStrategy | ||
import akka.stream.scaladsl.Sink | ||
import akka.stream.scaladsl.Source | ||
|
||
/** | ||
* Sources and sinks to integrate [[akka.actor.typed.pubsub.Topic]] with streams allowing for local or distributed | ||
* publishing and subscribing of elements through a stream. | ||
*/ | ||
object PubSub { | ||
|
||
/** | ||
* Create a source that will subscribe to a topic and stream messages published to the topic. Can be materialized | ||
* multiple times, each materialized stream will contain messages published after it was started. | ||
* | ||
* Note that it is not possible to propagate the backpressure from the running stream to the pub sub topic, | ||
* if the stream is backpressuring published messages are buffered up to a limit and if the limit is hit | ||
* the configurable `OverflowStrategy` decides what happens. It is not possible to use the `Backpressure` | ||
* strategy. | ||
* | ||
* @param topicActor The actor ref for an `akka.actor.typed.pubsub.Topic` actor representing a specific topic. | ||
* @param bufferSize The maximum number of messages to buffer if the stream applies backpressure | ||
* @param overflowStrategy Strategy to use once the buffer is full. | ||
* @tparam T The type of the published messages | ||
*/ | ||
@ApiMayChange | ||
def source[T]( | ||
topicActor: ActorRef[Topic.Command[T]], | ||
bufferSize: Int, | ||
overflowStrategy: OverflowStrategy): Source[T, NotUsed] = | ||
ActorSource | ||
.actorRef[T](PartialFunction.empty, PartialFunction.empty, bufferSize, overflowStrategy) | ||
.mapMaterializedValue { ref => | ||
topicActor ! Topic.Subscribe(ref) | ||
NotUsed | ||
} | ||
|
||
/** | ||
* Create a sink that will publish each message to the given topic. Note that there is no backpressure | ||
* from the topic, so care must be taken to not publish messages at a higher rate than that can be handled | ||
* by subscribers. If the topic does not have any subscribers when a message is published or the topic actor is stopped, | ||
* the message is sent to dead letters. | ||
* | ||
* @param topicActor The actor ref for an `akka.actor.typed.pubsub.Topic` actor representing a specific topic. | ||
* @tparam T the type of the messages that can be published | ||
*/ | ||
@ApiMayChange | ||
def sink[T](topicActor: ActorRef[Topic.Command[T]]): Sink[T, NotUsed] = { | ||
Sink | ||
.foreach[T] { message => | ||
topicActor ! Topic.Publish(message) | ||
} | ||
.mapMaterializedValue(_ => NotUsed) | ||
} | ||
} |
63 changes: 63 additions & 0 deletions
63
akka-stream-typed/src/test/scala/akka/stream/typed/scaladsl/PubSubSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,63 @@ | ||
/* | ||
* Copyright (C) 2021 Lightbend Inc. <https://www.lightbend.com> | ||
*/ | ||
|
||
/* | ||
* Copyright (C) 2009-2022 Lightbend Inc. <https://www.lightbend.com> | ||
*/ | ||
package akka.stream.typed.scaladsl | ||
|
||
import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit | ||
import akka.actor.typed.internal.pubsub.TopicImpl | ||
import akka.actor.typed.pubsub.Topic | ||
import akka.stream.OverflowStrategy | ||
import akka.stream.scaladsl.Source | ||
import akka.stream.testkit.scaladsl.TestSink | ||
import org.scalatest.wordspec.AnyWordSpecLike | ||
|
||
class PubSubSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike { | ||
|
||
"PubSub.source" should { | ||
|
||
"emit messages from the topic" in { | ||
val topic = testKit.spawn(Topic[String]("my-topic-1")) | ||
|
||
val source = PubSub.source(topic, 100, OverflowStrategy.fail) | ||
val sourceProbe = source.runWith(TestSink()) | ||
sourceProbe.ensureSubscription() | ||
|
||
// wait until subscription has been seen | ||
val probe = testKit.createTestProbe[TopicImpl.TopicStats]() | ||
probe.awaitAssert { | ||
topic ! TopicImpl.GetTopicStats(probe.ref) | ||
probe.expectMessageType[TopicImpl.TopicStats].localSubscriberCount should ===(1) | ||
} | ||
|
||
topic ! Topic.Publish("published") | ||
sourceProbe.requestNext("published") | ||
sourceProbe.cancel() | ||
} | ||
|
||
} | ||
|
||
"PubSub.sink" should { | ||
"publish messages" in { | ||
val topic = testKit.spawn(Topic[String]("my-topic-2")) | ||
|
||
val subscriberProbe = testKit.createTestProbe[String]() | ||
topic ! Topic.Subscribe(subscriberProbe.ref) | ||
|
||
// wait until subscription has been seen | ||
val probe = testKit.createTestProbe[TopicImpl.TopicStats]() | ||
probe.awaitAssert { | ||
topic ! TopicImpl.GetTopicStats(probe.ref) | ||
probe.expectMessageType[TopicImpl.TopicStats].localSubscriberCount should ===(1) | ||
} | ||
|
||
Source.single("published").runWith(PubSub.sink(topic)) | ||
|
||
subscriberProbe.expectMessage("published") | ||
} | ||
} | ||
|
||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters