chore: Remove Source#future in javadsl

Author: He-Pin

stream-tests/src/test/java/org/apache/pekko/stream/javadsl/LazyAndFutureSourcesTest.java

@@ -27,7 +27,6 @@
 import org.apache.pekko.testkit.PekkoSpec;
 import org.junit.ClassRule;
 import org.junit.Test;
-import scala.concurrent.Future;
 
 public class LazyAndFutureSourcesTest extends StreamTest {
 
@@ -41,14 +40,6 @@ public LazyAndFutureSourcesTest() {
 
   // note these are minimal happy path tests to cover API, more thorough tests are on the Scala side
 
-  @Test
-  public void future() throws Exception {
-    CompletionStage<List<String>> result =
-        Source.future(Future.successful("one")).runWith(Sink.seq(), system);
-
-    assertEquals(Arrays.asList("one"), result.toCompletableFuture().get(3, TimeUnit.SECONDS));
-  }
-
   @Test
   public void completionStage() throws Exception {
     CompletionStage<String> one = CompletableFuture.completedFuture("one");

stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala

@@ -56,6 +56,7 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
     ("apply" -> "fromIterator") ::
     ("apply" -> "fromFunctions") ::
     ("apply" -> "fromArray") ::
+    ("future" -> "completionStage") ::
     Nil
 
   // format: OFF
@@ -75,6 +76,7 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
       (classOf[scala.Function2[_, _, _]],                  classOf[java.util.function.BiFunction[_, _, _]]) :: // setup
       (classOf[scala.Function1[scala.Function1[_, _], _]], classOf[pekko.japi.function.Function2[_, _, _]]) ::
       (classOf[scala.concurrent.duration.FiniteDuration],  classOf[java.time.Duration]) ::
+      (classOf[scala.concurrent.Future[_]],                classOf[java.util.concurrent.CompletionStage[_]]) ::
       (classOf[pekko.stream.scaladsl.Source[_, _]],        classOf[pekko.stream.javadsl.Source[_, _]]) ::
       (classOf[pekko.stream.scaladsl.Sink[_, _]],          classOf[pekko.stream.javadsl.Sink[_, _]]) ::
       (classOf[pekko.stream.scaladsl.Flow[_, _, _]],       classOf[pekko.stream.javadsl.Flow[_, _, _]]) ::

stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes

@@ -202,3 +202,4 @@ ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.Acto
 ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.ActorMaterializer.this")
 ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.SetupStage")
 ProblemFilters.exclude[MissingClassProblem]("org.apache.pekko.stream.impl.SetupStage$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.stream.javadsl.Source.future")

stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala

@@ -20,7 +20,7 @@ import java.util.concurrent.{ CompletableFuture, CompletionStage }
 import scala.annotation.{ nowarn, varargs }
 import scala.annotation.unchecked.uncheckedVariance
 import scala.collection.immutable
-import scala.concurrent.{ Future, Promise }
+import scala.concurrent.Promise
 import scala.concurrent.ExecutionContext
 import scala.jdk.CollectionConverters._
 import scala.jdk.DurationConverters._
@@ -316,15 +316,6 @@ object Source {
   def failed[T](cause: Throwable): Source[T, NotUsed] =
     new Source(scaladsl.Source.failed(cause))
 
-  /**
-   * Emits a single value when the given Scala `Future` is successfully completed and then completes the stream.
-   * The stream fails if the `Future` is completed with a failure.
-   *
-   * Here for Java interoperability, the normal use from Java should be [[Source.completionStage]]
-   */
-  def future[T](futureElement: Future[T]): Source[T, NotUsed] =
-    scaladsl.Source.future(futureElement).asJava
-
   /**
    * Never emits any elements, never completes and never fails.
    * This stream could be useful in tests.
@@ -337,7 +328,7 @@ object Source {
    * If the `CompletionStage` is completed with a failure the stream is failed.
    */
   def completionStage[T](completionStage: CompletionStage[T]): Source[T, NotUsed] =
-    future(completionStage.asScala)
+    new Source(scaladsl.Source.future(completionStage.asScala))
 
   /**
    * Turn a `CompletionStage[Source]` into a source that will emit the values of the source when the future completes successfully.