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

=str #19128 #19127 rename Source/Sink factory apply overloads

Browse source
This commit is contained in:
Martynas Mickevičius 2015-12-17 11:48:30 +02:00
parent d7b45e0fc3
commit e6e476d82a
146 changed files with 910 additions and 740 deletions
Download patch file Download diff file Expand all files Collapse all files

32
akka-docs-dev/rst/java/code/docs/MigrationsJava.java
View file

@ -1,12 +1,16 @@
package docs;
import akka.actor.ActorSystem;
import akka.actor.Cancellable;
import akka.http.javadsl.model.Uri;
import akka.dispatch.Futures;
import akka.japi.function.Creator;
import akka.japi.Pair;
import akka.japi.function.Function;
import akka.stream.*;
import akka.stream.javadsl.*;
import akka.stream.testkit.TestPublisher;
import akka.stream.testkit.TestSubscriber;
import akka.util.ByteString;
import scala.Option;
import scala.concurrent.Future;
@ -14,6 +18,9 @@ import scala.concurrent.duration.FiniteDuration;
import scala.concurrent.Promise;
import scala.runtime.BoxedUnit;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
@ -25,6 +32,7 @@ public class MigrationsJava {
// This is compile-only code, no need for actually running anything.
public static ActorMaterializer mat = null;
public static ActorSystem sys = null;
public static class SomeInputStream extends InputStream {
public SomeInputStream() {}
@ -133,12 +141,34 @@ public class MigrationsJava {
// Complete the promise with an empty option to emulate the old lazyEmpty
promise.trySuccess(scala.Option.empty());
final Source<String, Cancellable> sourceUnderTest = Source.tick(
final Source<String, Cancellable> ticks = Source.tick(
FiniteDuration.create(0, TimeUnit.MILLISECONDS),
FiniteDuration.create(200, TimeUnit.MILLISECONDS),
"tick");
final Source<Integer, BoxedUnit> pubSource =
Source.fromPublisher(TestPublisher.<Integer>manualProbe(true, sys));
final Source<Integer, BoxedUnit> futSource =
Source.fromFuture(Futures.successful(42));
final Source<Integer, Subscriber<Integer>> subSource =
Source.<Integer>asSubscriber();
//#source-creators
//#sink-creators
final Sink<Integer, BoxedUnit> subSink =
Sink.fromSubscriber(TestSubscriber.<Integer>manualProbe(sys));
//#sink-creators
//#sink-as-publisher
final Sink<Integer, Publisher<Integer>> pubSink =
Sink.<Integer>asPublisher(false);
final Sink<Integer, Publisher<Integer>> pubSinkFanout =
Sink.<Integer>asPublisher(true);
//#sink-as-publisher
//#empty-flow
Flow<Integer, Integer, BoxedUnit> emptyFlow = Flow.<Integer>create();
// or

72
akka-docs-dev/rst/java/migration-guide-1.0-2.x-java.rst
View file

@ -224,13 +224,14 @@ should be replaced by
Source constructor name changes
===============================
``Source.lazyEmpty`` have been replaced by ``Source.maybe`` which returns a ``Promise`` that can be completed by one or
``Source.lazyEmpty`` has been replaced by ``Source.maybe`` which returns a ``Promise`` that can be completed by one or
zero elements by providing an ``Option``. This is different from ``lazyEmpty`` which only allowed completion to be
sent, but no elements.
The ``from()`` overload on ``Source`` that provide a tick source (``Source.from(delay,interval,tick)``)
is replaced by the named method ``Source.tick()`` to reduce the number of overloads and to make the function more
discoverable.
The ``from()`` overload on ``Source`` has been refactored to separate methods to reduce the number of overloads and
make source creation more discoverable.
``Source.subscriber`` has been renamed to ``Source.asSubscriber``.
Update procedure
----------------
@ -238,6 +239,9 @@ Update procedure
1. All uses of ``Source.lazyEmpty`` should be replaced by ``Source.maybe`` and the returned ``Promise`` completed with
a ``None`` (an empty ``Option``)
2. Replace all uses of ``Source.from(delay,interval,tick)`` with the method ``Source.tick(delay,interval,tick)``
3. Replace all uses of ``Source.from(publisher)`` with the method ``Source.fromPublisher(publisher)``
4. Replace all uses of ``Source.from(future)`` with the method ``Source.fromFuture(future))``
5. Replace all uses of ``Source.subscriber`` with the method ``Source.asSubscriber``
Example
^^^^^^^
@ -250,15 +254,51 @@ Example
promise.trySuccess(BoxedUnit.UNIT);
// This no longer works!
final Source<String, Cancellable> sourceUnderTest = Source.from(
final Source<String, Cancellable> ticks = Source.from(
FiniteDuration.create(0, TimeUnit.MILLISECONDS),
FiniteDuration.create(200, TimeUnit.MILLISECONDS),
"tick");
// This no longer works!
final Source<Integer, BoxedUnit> pubSource =
Source.from(TestPublisher.<Integer>manualProbe(true, sys));
// This no longer works!
final Source<Integer, BoxedUnit> futSource =
Source.from(Futures.successful(42));
// This no longer works!
final Source<Integer, Subscriber<Integer>> subSource =
Source.<Integer>subscriber();
should be replaced by
.. includecode:: code/docs/MigrationsJava.java#source-creators
Sink constructor name changes
=============================
``Sink.create(subscriber)`` has been renamed to ``Sink.fromSubscriber(subscriber)`` to reduce the number of overloads and
make sink creation more discoverable.
Update procedure
----------------
1. Replace all uses of ``Sink.create(subscriber)`` with the method ``Sink.fromSubscriber(subscriber)``
Example
^^^^^^^
::
// This no longer works!
final Sink<Integer, BoxedUnit> subSink =
Sink.create(TestSubscriber.<Integer>manualProbe(sys));
should be replaced by
.. includecode:: code/docs/MigrationsJava.java#sink-creators
``Flow.empty()`` have been removed
==================================
@ -308,16 +348,30 @@ should be replaced by
It was a common user mistake to use ``Sink.publisher`` and get into trouble since it would only support
a single ``Subscriber``, and the discoverability of the apprpriate fix was non-obvious (Sink.fanoutPublisher).
To make the decision whether to support fanout or not an active one, the aforementioned methods have been
replaced with a single method: ``Sink.publisher(fanout: Boolean)``.
replaced with a single method: ``Sink.asPublisher(fanout: Boolean)``.
Update procedure
----------------
1. Replace all occurences of ``Sink.publisher`` with ``Sink.publisher(false)``
2. Replace all occurences of ``Sink.fanoutPublisher`` with ``Sink.publisher(true)``
1. Replace all occurences of ``Sink.publisher`` with ``Sink.asPublisher(false)``
2. Replace all occurences of ``Sink.fanoutPublisher`` with ``Sink.asPublisher(true)``
TODO: code example
Example
^^^^^^^
::
// This no longer works!
final Sink<Integer, Publisher<Integer>> pubSink =
Sink.<Integer>publisher();
// This no longer works!
final Sink<Integer, Publisher<Integer>> pubSink =
Sink.<Integer>fanoutPublisher(2, 8);
should be replaced by
.. includecode:: code/docs/MigrationsJava.java#sink-as-publisher
FlexiMerge an FlexiRoute has been replaced by GraphStage
========================================================

14
akka-docs-dev/rst/java/stream-integrations.rst
View file

@ -8,7 +8,7 @@ Integrating with Actors
=======================
For piping the elements of a stream as messages to an ordinary actor you can use the
``Sink.actorRef``. Messages can be sent to a stream via the :class:`ActorRef` that is
``Sink.actorRef``. Messages can be sent to a stream via the :class:`ActorRef` that is
materialized by ``Source.actorRef``.
For more advanced use cases the :class:`ActorPublisher` and :class:`ActorSubscriber` traits are
@ -32,8 +32,8 @@ Akka Streams :class:`Source` or :class:`Sink`.
Source.actorRef
^^^^^^^^^^^^^^^
Messages sent to the actor that is materialized by ``Source.actorRef`` will be emitted to the
stream if there is demand from downstream, otherwise they will be buffered until request for
Messages sent to the actor that is materialized by ``Source.actorRef`` will be emitted to the
stream if there is demand from downstream, otherwise they will be buffered until request for
demand is received.
Depending on the defined :class:`OverflowStrategy` it might drop elements if there is no space
@ -44,7 +44,7 @@ actor interface.
The stream can be completed successfully by sending ``akka.actor.PoisonPill`` or
``akka.actor.Status.Success`` to the actor reference.
The stream can be completed with failure by sending ``akka.actor.Status.Failure`` to the
The stream can be completed with failure by sending ``akka.actor.Status.Failure`` to the
actor reference.
The actor will be stopped when the stream is completed, failed or cancelled from downstream,
@ -108,12 +108,12 @@ This is how it can be used as input :class:`Source` to a :class:`Flow`:
.. includecode:: ../../../akka-samples/akka-docs-java-lambda/src/test/java/docs/stream/ActorPublisherDocTest.java#actor-publisher-usage
You can only attach one subscriber to this publisher. Use a ``Broadcast``-element or
attach a ``Sink.publisher(true)`` to enable multiple subscribers.
attach a ``Sink.asPublisher(true)`` to enable multiple subscribers.
ActorSubscriber
^^^^^^^^^^^^^^^
Extend :class:`akka.stream.actor.AbstractActorSubscriber` to make your class a stream subscriber with
Extend :class:`akka.stream.actor.AbstractActorSubscriber` to make your class a stream subscriber with
full control of stream back pressure. It will receive
``ActorSubscriberMessage.OnNext``, ``ActorSubscriberMessage.OnComplete`` and ``ActorSubscriberMessage.OnError``
messages from the stream. It can also receive other, non-stream messages, in the same way as any actor.
@ -414,7 +414,7 @@ by using the Publisher-:class:`Sink`:
.. includecode:: ../../../akka-samples/akka-docs-java-lambda/src/test/java/docs/stream/ReactiveStreamsDocTest.java#source-publisher
A publisher that is created with ``Sink.publisher(false)`` supports only a single subscription.
A publisher that is created with ``Sink.asPublisher(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: