Typed stream adapters, #23604

* Change more general factories to private
* Typed Streams docs
* Remove BoxedUnit from Java Api
* Use JavaPartialFunction in Java examples
* Doc wording improvements, formatting fixes, no verification diagrams

akka-stream-typed/src/main/scala/akka/stream/typed/ActorMaterializer.scala

@@ -1,3 +1,6 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
 package akka.stream.typed
 
 import akka.actor.typed.ActorSystem

akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/ActorSink.scala

@@ -1,11 +1,11 @@
-/*
- * Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com/>
-  */
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
 package akka.stream.typed.javadsl
 
 import akka.actor.typed._
 import akka.NotUsed
-import akka.stream.scaladsl._
+import akka.stream.javadsl._
 import akka.stream.typed
 
 /**
@@ -29,7 +29,7 @@ object ActorSink {
    * limiting stage in front of this `Sink`.
    */
   def actorRef[T](ref: ActorRef[T], onCompleteMessage: T, onFailureMessage: akka.japi.function.Function[Throwable, T]): Sink[T, NotUsed] =
-    typed.scaladsl.ActorSink.actorRef(ref, onCompleteMessage, onFailureMessage.apply)
+    typed.scaladsl.ActorSink.actorRef(ref, onCompleteMessage, onFailureMessage.apply).asJava
 
   /**
    * Sends the elements of the stream to the given `ActorRef` that sends back back-pressure signal.
@@ -52,6 +52,6 @@ object ActorSink {
     onCompleteMessage: M,
     onFailureMessage:  akka.japi.function.Function[Throwable, M]): Sink[T, NotUsed] =
     typed.scaladsl.ActorSink.actorRefWithAck(
-      ref, messageAdapter.apply, onInitMessage.apply, ackMessage, onCompleteMessage, onFailureMessage.apply)
+      ref, messageAdapter.apply, onInitMessage.apply, ackMessage, onCompleteMessage, onFailureMessage.apply).asJava
 
 }

akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/ActorSource.scala

@@ -1,9 +1,11 @@
-/*
- * Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com/>
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
  */
 
 package akka.stream.typed.javadsl
 
+import java.util.function.Predicate
+
 import akka.actor.typed._
 import akka.stream.OverflowStrategy
 import akka.stream.javadsl._
@@ -46,11 +48,11 @@ object ActorSource {
    * @param overflowStrategy Strategy that is used when incoming elements cannot fit inside the buffer
    */
   def actorRef[T](
-    completionMatcher: PartialFunction[T, Unit],
+    completionMatcher: Predicate[T],
     failureMatcher:    PartialFunction[T, Throwable],
     bufferSize:        Int, overflowStrategy: OverflowStrategy): Source[T, ActorRef[T]] = {
     akka.stream.typed.scaladsl.ActorSource.actorRef(
-      completionMatcher, failureMatcher,
-      bufferSize, overflowStrategy).asJava
+      { case m if completionMatcher.test(m) ⇒ }: PartialFunction[T, Unit],
+      failureMatcher, bufferSize, overflowStrategy).asJava
   }
 }

akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSink.scala

@@ -1,5 +1,5 @@
-/*
- * Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com/>
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
  */
 package akka.stream.typed.scaladsl
 

akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSource.scala

@@ -1,5 +1,5 @@
-/*
- * Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com/>
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
  */
 
 package akka.stream.typed.scaladsl

akka-stream-typed/src/test/java/akka/stream/typed/javadsl/ActorSourceSinkCompileTest.java

@@ -1,14 +1,16 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
+
 package akka.stream.typed.javadsl;
 
 import akka.actor.typed.ActorRef;
 import akka.actor.typed.ActorSystem;
+import akka.japi.JavaPartialFunction;
 import akka.stream.ActorMaterializer;
 import akka.stream.OverflowStrategy;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
-import scala.PartialFunction$;
-import scala.runtime.AbstractPartialFunction;
-import scala.runtime.BoxedUnit;
 
 public class ActorSourceSinkCompileTest {
 
@@ -47,38 +49,36 @@ public class ActorSourceSinkCompileTest {
   }
 
   {
-    final AbstractPartialFunction<String, BoxedUnit> completionMatcher = new AbstractPartialFunction<String, BoxedUnit>() {
-      @Override
-      public boolean isDefinedAt(String s) {
-        return s == "complete";
-      }
-    };
-
     ActorSource
       .actorRef(
-        completionMatcher,
-        PartialFunction$.MODULE$.empty(), // FIXME make the API nicer
+        (m) -> m == "complete",
+        new JavaPartialFunction<String, Throwable>() {
+          @Override
+          public Throwable apply(String x, boolean isCheck) throws Exception {
+            throw noMatch();
+          }
+        },
         10,
         OverflowStrategy.dropBuffer())
       .to(Sink.seq());
   }
 
   {
-    final AbstractPartialFunction<Protocol, Throwable> failureMatcher = new AbstractPartialFunction<Protocol, Throwable>() {
+    final JavaPartialFunction<Protocol, Throwable> failureMatcher = new JavaPartialFunction<Protocol, Throwable>() {
       @Override
-      public boolean isDefinedAt(Protocol p) {
-        return p instanceof Failure;
-      }
-
-      @Override
-      public Throwable apply(Protocol p) {
-        return ((Failure)p).ex;
+      public Throwable apply(Protocol p, boolean isCheck) throws Exception {
+        if (p instanceof Failure) {
+          return ((Failure)p).ex;
+        }
+        else {
+          throw noMatch();
+        }
       }
     };
 
     ActorSource
       .actorRef(
-        PartialFunction$.MODULE$.empty(), // FIXME make the API nicer
+        (m) -> false,
         failureMatcher, 10,
         OverflowStrategy.dropBuffer())
       .to(Sink.seq());

akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSinkExample.java

@@ -0,0 +1,53 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
+
+package docs.akka.stream.typed;
+
+// #actor-sink-ref
+import akka.NotUsed;
+import akka.actor.typed.ActorRef;
+import akka.stream.ActorMaterializer;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.stream.typed.javadsl.ActorSink;
+// #actor-sink-ref
+
+public class ActorSinkExample {
+
+  // #actor-sink-ref
+
+  interface Protocol {}
+  class Message implements Protocol {
+    private final String msg;
+    public Message(String msg) {
+      this.msg = msg;
+    }
+  }
+  class Complete implements Protocol {}
+  class Fail implements Protocol {
+    private final Throwable ex;
+    public Fail(Throwable ex) {
+      this.ex = ex;
+    }
+  }
+  // #actor-sink-ref
+
+  final ActorMaterializer mat = null;
+
+  {
+    // #actor-sink-ref
+
+    final ActorRef<Protocol> actor = null;
+
+    final Sink<Protocol, NotUsed> sink = ActorSink.actorRef(
+      actor,
+      new Complete(),
+      Fail::new
+    );
+
+    Source.<Protocol>single(new Message("msg1")).runWith(sink, mat);
+    // #actor-sink-ref
+  }
+
+}

akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSinkWithAckExample.java

@@ -0,0 +1,66 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
+
+package docs.akka.stream.typed;
+
+// #actor-sink-ref-with-ack
+import akka.NotUsed;
+import akka.actor.typed.ActorRef;
+import akka.stream.ActorMaterializer;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.stream.typed.javadsl.ActorSink;
+// #actor-sink-ref-with-ack
+
+public class ActorSinkWithAckExample {
+
+  // #actor-sink-ref-with-ack
+
+  class Ack {}
+
+  interface Protocol {}
+  class Init implements Protocol {
+    private final ActorRef<Ack> ack;
+    public Init(ActorRef<Ack> ack) {
+      this.ack = ack;
+    }
+  }
+  class Message implements Protocol {
+    private final ActorRef<Ack> ackTo;
+    private final String msg;
+    public Message(ActorRef<Ack> ackTo, String msg) {
+      this.ackTo = ackTo;
+      this.msg = msg;
+    }
+  }
+  class Complete implements Protocol {}
+  class Fail implements Protocol {
+    private final Throwable ex;
+    public Fail(Throwable ex) {
+      this.ex = ex;
+    }
+  }
+  // #actor-sink-ref-with-ack
+
+  final ActorMaterializer mat = null;
+
+  {
+    // #actor-sink-ref-with-ack
+
+    final ActorRef<Protocol> actor = null;
+
+    final Sink<String, NotUsed> sink = ActorSink.actorRefWithAck(
+      actor,
+      Message::new,
+      Init::new,
+      new Ack(),
+      new Complete(),
+      Fail::new
+    );
+
+    Source.single("msg1").runWith(sink, mat);
+    // #actor-sink-ref-with-ack
+  }
+
+}

akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSourceExample.java

@@ -0,0 +1,74 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
+
+package docs.akka.stream.typed;
+
+// #actor-source-ref
+import akka.actor.typed.ActorRef;
+import akka.japi.JavaPartialFunction;
+import akka.stream.ActorMaterializer;
+import akka.stream.OverflowStrategy;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.stream.typed.javadsl.ActorSource;
+// #actor-source-ref
+
+public class ActorSourceExample {
+
+  // #actor-source-ref
+
+  interface Protocol {}
+  class Message implements Protocol {
+    private final String msg;
+    public Message(String msg) {
+      this.msg = msg;
+    }
+  }
+  class Complete implements Protocol {}
+  class Fail implements Protocol {
+    private final Exception ex;
+    public Fail(Exception ex) {
+      this.ex = ex;
+    }
+  }
+  // #actor-source-ref
+
+  final ActorMaterializer mat = null;
+
+  {
+    // #actor-source-ref
+
+    final JavaPartialFunction<Protocol, Throwable> failureMatcher =
+      new JavaPartialFunction<Protocol, Throwable>() {
+        public Throwable apply(Protocol p, boolean isCheck) {
+          if (p instanceof Fail) {
+            return ((Fail)p).ex;
+          } else {
+            throw noMatch();
+          }
+        }
+      };
+
+    final Source<Protocol, ActorRef<Protocol>> source = ActorSource.actorRef(
+      (m) -> m instanceof Complete,
+      failureMatcher,
+      8,
+      OverflowStrategy.fail()
+    );
+
+    final ActorRef<Protocol> ref = source.collect(new JavaPartialFunction<Protocol, String>() {
+      public String apply(Protocol p, boolean isCheck) {
+        if (p instanceof Message) {
+          return ((Message)p).msg;
+       } else {
+          throw noMatch();
+       }
+     }
+    }).to(Sink.foreach(System.out::println)).run(mat);
+
+    ref.tell(new Message("msg1"));
+    // ref.tell("msg2"); Does not compile
+    // #actor-source-ref
+  }
+}

akka-stream-typed/src/test/scala/akka/stream/typed/scaladsl/ActorSourceSinkSpec.scala

@@ -1,9 +1,9 @@
-/*
- * Copyright (C) 2017 Lightbend Inc. <http://www.lightbend.com/>
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
  */
 package akka.stream.typed.scaladsl
 
-import akka.actor.typed.scaladsl.Actor
+import akka.actor.typed.scaladsl.Behaviors
 import akka.stream.OverflowStrategy
 import akka.actor.typed.{ ActorRef, ActorSystem }
 import akka.testkit.TestKit
@@ -59,20 +59,20 @@ class ActorSourceSinkSpec extends TestKit(akka.actor.ActorSystem("ActorSourceSin
     "obey protocol" in {
       val p = TestProbe[AckProto]()
 
-      val autoPilot = Actor.immutable[AckProto] {
+      val autoPilot = Behaviors.immutable[AckProto] {
         (ctx, msg) ⇒
           msg match {
             case m @ Init(sender) ⇒
               p.ref ! m
               sender ! "ACK"
-              Actor.same
+              Behaviors.same
             case m @ Msg(sender, _) ⇒
               p.ref ! m
               sender ! "ACK"
-              Actor.same
+              Behaviors.same
             case m ⇒
               p.ref ! m
-              Actor.same
+              Behaviors.same
           }
       }
 

akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala

@@ -0,0 +1,98 @@
+/**
+ * Copyright (C) 2018 Lightbend Inc. <https://www.lightbend.com/>
+ */
+
+package docs.akka.stream.typed
+
+import akka.NotUsed
+import akka.stream.ActorMaterializer
+
+object ActorSourceSinkExample {
+
+  implicit val mat: ActorMaterializer = ???
+
+  {
+    // #actor-source-ref
+    import akka.actor.typed.ActorRef
+    import akka.stream.OverflowStrategy
+    import akka.stream.scaladsl.{ Sink, Source }
+    import akka.stream.typed.scaladsl.ActorSource
+
+    trait Protocol
+    case class Message(msg: String) extends Protocol
+    case object Complete extends Protocol
+    case class Fail(ex: Exception) extends Protocol
+
+    val source: Source[Protocol, ActorRef[Protocol]] = ActorSource.actorRef[Protocol](
+      completionMatcher = {
+        case Complete ⇒
+      },
+      failureMatcher = {
+        case Fail(ex) ⇒ ex
+      },
+      bufferSize = 8,
+      overflowStrategy = OverflowStrategy.fail
+    )
+
+    val ref = source.collect {
+      case Message(msg) ⇒ msg
+    }.to(Sink.foreach(println)).run()
+
+    ref ! Message("msg1")
+    // ref ! "msg2" Does not compile
+    // #actor-source-ref
+  }
+
+  {
+    // #actor-sink-ref
+    import akka.actor.typed.ActorRef
+    import akka.stream.scaladsl.{ Sink, Source }
+    import akka.stream.typed.scaladsl.ActorSink
+
+    trait Protocol
+    case class Message(msg: String) extends Protocol
+    case object Complete extends Protocol
+    case class Fail(ex: Throwable) extends Protocol
+
+    val actor: ActorRef[Protocol] = ???
+
+    val sink: Sink[Protocol, NotUsed] = ActorSink.actorRef[Protocol](
+      ref = actor,
+      onCompleteMessage = Complete,
+      onFailureMessage = Fail.apply
+    )
+
+    Source.single(Message("msg1")).runWith(sink)
+    // #actor-sink-ref
+  }
+
+  {
+    // #actor-sink-ref-with-ack
+    import akka.actor.typed.ActorRef
+    import akka.stream.scaladsl.{ Sink, Source }
+    import akka.stream.typed.scaladsl.ActorSink
+
+    trait Ack
+    object Ack extends Ack
+
+    trait Protocol
+    case class Init(ackTo: ActorRef[Ack]) extends Protocol
+    case class Message(ackTo: ActorRef[Ack], msg: String) extends Protocol
+    case object Complete extends Protocol
+    case class Fail(ex: Throwable) extends Protocol
+
+    val actor: ActorRef[Protocol] = ???
+
+    val sink: Sink[String, NotUsed] = ActorSink.actorRefWithAck(
+      ref = actor,
+      onCompleteMessage = Complete,
+      onFailureMessage = Fail.apply,
+      messageAdapter = Message.apply,
+      onInitMessage = Init.apply,
+      ackMessage = Ack
+    )
+
+    Source.single("msg1").runWith(sink)
+    // #actor-sink-ref-with-ack
+  }
+}