=str typos in returned types in javadsl
sadly a reflection based spec cannot easily test this
This commit is contained in:
parent
81bc5c76bc
commit
122daaa88d
3 changed files with 57 additions and 3 deletions
|
|
@ -75,7 +75,6 @@ public class FlowTest {
|
|||
}, system.dispatcher());
|
||||
|
||||
probe.expectMsgEquals("de");
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
|
|
@ -0,0 +1,55 @@
|
|||
/**
|
||||
* Copyright (C) 2014 Typesafe Inc. <http://www.typesafe.com>
|
||||
*/
|
||||
package akka.stream.javadsl;
|
||||
|
||||
import akka.actor.ActorRef;
|
||||
import akka.actor.ActorSystem;
|
||||
import akka.dispatch.Foreach;
|
||||
import akka.dispatch.Futures;
|
||||
import akka.dispatch.OnSuccess;
|
||||
import akka.japi.Pair;
|
||||
import akka.japi.Util;
|
||||
import akka.stream.FlowMaterializer;
|
||||
import akka.stream.MaterializerSettings;
|
||||
import akka.stream.OverflowStrategy;
|
||||
import akka.stream.Transformer;
|
||||
import akka.stream.javadsl.japi.*;
|
||||
import akka.stream.testkit.AkkaSpec;
|
||||
import akka.testkit.JavaTestKit;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Test;
|
||||
import org.reactivestreams.Publisher;
|
||||
import scala.Option;
|
||||
import scala.collection.immutable.Seq;
|
||||
import scala.concurrent.Await;
|
||||
import scala.concurrent.Future;
|
||||
import scala.concurrent.duration.Duration;
|
||||
import scala.concurrent.duration.FiniteDuration;
|
||||
import scala.runtime.BoxedUnit;
|
||||
import scala.util.Try;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
public class SinkTest {
|
||||
|
||||
@ClassRule
|
||||
public static AkkaJUnitActorSystemResource actorSystemResource = new AkkaJUnitActorSystemResource("FlowTest",
|
||||
AkkaSpec.testConf());
|
||||
|
||||
final ActorSystem system = actorSystemResource.getSystem();
|
||||
|
||||
final MaterializerSettings settings = new MaterializerSettings(2, 4, 2, 4, "akka.test.stream-dispatcher");
|
||||
final FlowMaterializer materializer = FlowMaterializer.create(settings, system);
|
||||
|
||||
@Test
|
||||
public void mustBeAbleToUseFanoutPublisher() throws Exception {
|
||||
KeyedSink<Object, Publisher<Object>> pubSink = Sink.fanoutPublisher(2, 2);
|
||||
Publisher<Object> publisher = Source.from(new ArrayList<Object>()).runWith(pubSink, materializer);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -92,7 +92,7 @@ object Sink {
|
|||
* A `Sink` that materializes into a [[org.reactivestreams.Publisher]]
|
||||
* that can handle more than one [[org.reactivestreams.Subscriber]].
|
||||
*/
|
||||
def fanoutPublisher[T](initialBufferSize: Int, maximumBufferSize: Int): KeyedSink[Publisher[T], T] =
|
||||
def fanoutPublisher[T](initialBufferSize: Int, maximumBufferSize: Int): KeyedSink[T, Publisher[T]] =
|
||||
new KeyedSink(scaladsl.Sink.fanoutPublisher(initialBufferSize, maximumBufferSize))
|
||||
|
||||
/**
|
||||
|
|
@ -116,7 +116,7 @@ object Sink {
|
|||
* function evaluation when the input stream ends, or completed with `Failure`
|
||||
* if there is an error is signaled in the stream.
|
||||
*/
|
||||
def fold[U, T](zero: U, f: Function[akka.japi.Pair[U, T], U]): KeyedSink[T, U] = {
|
||||
def fold[U, T](zero: U, f: Function[akka.japi.Pair[U, T], U]): KeyedSink[T, Future[U]] = {
|
||||
val sSink = scaladsl.Sink.fold[U, T](zero) { case (a, b) ⇒ f.apply(akka.japi.Pair(a, b)) }
|
||||
new KeyedSink(sSink)
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue