raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

#19522 make javadsl.Sink.asPublisher nicer

Browse source
This commit is contained in:
Roland Kuhn 2016-01-20 21:01:27 +01:00
parent b1351b36ed
commit 9427052fd0
12 changed files with 43 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

8
akka-docs/rst/java/code/docs/stream/MigrationsJava.java
View file

@ -7,6 +7,9 @@ import java.util.stream.Stream;
import akka.japi.Pair;
import akka.stream.javadsl.*;
//#asPublisher-import
import static akka.stream.javadsl.AsPublisher.*;
//#asPublisher-import
public class MigrationsJava {
@ -19,6 +22,11 @@ public class MigrationsJava {
Stream.iterate(new Pair<>(in, 0),
p -> new Pair<>(in, p.second() + 1)).iterator());
//#expand-state
//#asPublisher
Sink.asPublisher(WITH_FANOUT); // instead of Sink.asPublisher(true)
Sink.asPublisher(WITHOUT_FANOUT); // instead of Sink.asPublisher(false)
//#asPublisher
}
}

8
akka-docs/rst/java/code/docs/stream/ReactiveStreamsDocTest.java
View file

@ -83,7 +83,7 @@ public class ReactiveStreamsDocTest {
final Fixture.RS rs = new Fixture.RS() {
@Override
public Publisher<Tweet> tweets() {
return TwitterStreamQuickstartDocTest.Model.tweets.runWith(Sink.asPublisher(false), mat);
return TwitterStreamQuickstartDocTest.Model.tweets.runWith(Sink.asPublisher(AsPublisher.WITHOUT_FANOUT), mat);
}
/**
@ -177,7 +177,9 @@ public class ReactiveStreamsDocTest {
{
//#source-publisher
final Publisher<Author> authorPublisher =
Source.fromPublisher(rs.tweets()).via(authors).runWith(Sink.asPublisher(false), mat);
Source.fromPublisher(rs.tweets())
.via(authors)
.runWith(Sink.asPublisher(AsPublisher.WITHOUT_FANOUT), mat);
authorPublisher.subscribe(rs.storage());
//#source-publisher
@ -197,7 +199,7 @@ public class ReactiveStreamsDocTest {
final Publisher<Author> authorPublisher =
Source.fromPublisher(rs.tweets())
.via(authors)
.runWith(Sink.asPublisher(true), mat);
.runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), mat);
authorPublisher.subscribe(rs.storage());
authorPublisher.subscribe(rs.alert());

12
akka-docs/rst/java/stream/migration-guide-2.0-2.4-java.rst
View file

@ -73,3 +73,15 @@ In Akka 2.4.x this is formulated like so:
.. includecode:: ../code/docs/stream/MigrationsJava.java#expand-state
Changed Sinks
=============
Sink.asPublisher is now configured using an enum
------------------------------------------------
In order to not use a meaningless boolean parameter we have changed the signature to:
.. includecode:: ../code/docs/stream/MigrationsJava.java#asPublisher-import
.. includecode:: ../code/docs/stream/MigrationsJava.java#asPublisher

4
akka-docs/rst/java/stream/stream-integrations.rst
View file

@ -108,7 +108,7 @@ This is how it can be used as input :class:`Source` to a :class:`Flow`:
.. includecode:: ../code/docs/stream/ActorPublisherDocTest.java#actor-publisher-usage
You can only attach one subscriber to this publisher. Use a ``Broadcast``-element or
attach a ``Sink.asPublisher(true)`` to enable multiple subscribers.
attach a ``Sink.asPublisher(AsPublisher.WITH_FANOUT)`` to enable multiple subscribers.
ActorSubscriber
^^^^^^^^^^^^^^^
@ -414,7 +414,7 @@ by using the Publisher-:class:`Sink`:
.. includecode:: ../code/docs/stream/ReactiveStreamsDocTest.java#source-publisher
A publisher that is created with ``Sink.asPublisher(false)`` supports only a single subscription.
A publisher that is created with ``Sink.asPublisher(AsPublisher.WITHOUT_FANOUT)`` supports only a single subscription.
Additional subscription attempts will be rejected with an :class:`IllegalStateException`.
A publisher that supports multiple subscribers using fan-out/broadcasting is created as follows:

2
akka-docs/rst/scala/code/docs/stream/ReactiveStreamsDocSpec.scala
View file

@ -42,7 +42,7 @@ class ReactiveStreamsDocSpec extends AkkaSpec {
val impl = new Fixture {
override def tweets: Publisher[Tweet] =
TwitterStreamQuickstartDocSpec.tweets.runWith(Sink.asPublisher(false))
TwitterStreamQuickstartDocSpec.tweets.runWith(Sink.asPublisher(fanout = false))
override def storage = TestSubscriber.manualProbe[Author]

2
akka-docs/rst/scala/stream/stream-integrations.rst
View file

@ -409,7 +409,7 @@ by using the Publisher-:class:`Sink`:
.. includecode:: ../code/docs/stream/ReactiveStreamsDocSpec.scala#source-publisher
A publisher that is created with ``Sink.asPublisher(false)`` supports only a single subscription.
A publisher that is created with ``Sink.asPublisher(fanout = false)`` supports only a single subscription.
Additional subscription attempts will be rejected with an :class:`IllegalStateException`.
A publisher that supports multiple subscribers using fan-out/broadcasting is created as follows: