Skip to content

chore: Remove Source#future in javadsl #2554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
He-Pin wants to merge 1 commit into
base: main
Choose a base branch
from
Open

chore: Remove Source#future in javadsl #2554

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 0 additions & 9 deletions stream-tests/src/test/java/org/apache/pekko/stream/javadsl/LazyAndFutureSourcesTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand All @@ -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");
Expand Down
2 changes: 2 additions & 0 deletions stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
("apply" -> "fromIterator") ::
("apply" -> "fromFunctions") ::
("apply" -> "fromArray") ::
("future" -> "completionStage") ::
Nil

// format: OFF
Expand All @@ -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[_, _, _]]) ::
Expand Down
1 change: 1 addition & 0 deletions stream/src/main/mima-filters/2.0.x.backwards.excludes/remove-deprecated-methods.excludes
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
13 changes: 2 additions & 11 deletions stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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.
Expand All @@ -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.
Expand Down