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

Move ConstantFun to akka.util, #22813

Browse source
This commit is contained in:
Hawstein 2017-05-11 00:00:42 +08:00 • committed by Patrik Nordwall
parent f064d1321a
commit b3d00afdd1
18 changed files with 120 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

47
akka-actor/src/main/scala/akka/util/ConstantFun.scala Normal file
View file

@ -0,0 +1,47 @@
/**
* Copyright (C) 2015-2017 Lightbend Inc. <http://www.lightbend.com>
*/
package akka.util
import akka.annotation.InternalApi
import akka.japi.function.{ Function ⇒ JFun, Function2 ⇒ JFun2 }
import akka.japi.{ Pair ⇒ JPair }
/**
* INTERNAL API
*/
@InternalApi private[akka] object ConstantFun {
private[this] val JavaIdentityFunction = new JFun[Any, Any] {
@throws(classOf[Exception]) override def apply(param: Any): Any = param
}
val JavaPairFunction = new JFun2[AnyRef, AnyRef, AnyRef JPair AnyRef] {
def apply(p1: AnyRef, p2: AnyRef): AnyRef JPair AnyRef = JPair(p1, p2)
}
def javaCreatePairFunction[A, B]: JFun2[A, B, JPair[A, B]] = JavaPairFunction.asInstanceOf[JFun2[A, B, JPair[A, B]]]
def javaIdentityFunction[T]: JFun[T, T] = JavaIdentityFunction.asInstanceOf[JFun[T, T]]
def scalaIdentityFunction[T]: T ⇒ T = conforms.asInstanceOf[Function[T, T]]
def scalaAnyToNone[A, B]: A ⇒ Option[B] = none
def scalaAnyTwoToNone[A, B, C]: (A, B) ⇒ Option[C] = two2none
def javaAnyToNone[A, B]: A ⇒ Option[B] = none
def nullFun[T] = _nullFun.asInstanceOf[Any ⇒ T]
val zeroLong = (_: Any) ⇒ 0L
val oneLong = (_: Any) ⇒ 1L
val oneInt = (_: Any) ⇒ 1
private val _nullFun = (_: Any) ⇒ null
private val conforms = (a: Any) ⇒ a
private val none = (_: Any) ⇒ None
private val two2none = (_: Any, _: Any) ⇒ None
}

2
akka-bench-jmh/src/main/scala/akka/stream/MaterializationBenchmark.scala
View file

@ -8,11 +8,11 @@ import java.util.concurrent.TimeUnit
import akka.NotUsed import akka.NotUsed
import akka.actor.ActorSystem import akka.actor.ActorSystem
import akka.stream.scaladsl._ import akka.stream.scaladsl._
import akka.util.ConstantFun
import org.openjdk.jmh.annotations._ import org.openjdk.jmh.annotations._
import scala.concurrent.Await import scala.concurrent.Await
import scala.concurrent.duration._ import scala.concurrent.duration._
import scala.concurrent.Future import scala.concurrent.Future
import akka.stream.impl.ConstantFun
import akka.Done import akka.Done
object MaterializationBenchmark { object MaterializationBenchmark {

2
akka-stream-tests-tck/src/test/scala/akka/stream/tck/FlattenTest.scala
View file

@ -3,9 +3,9 @@
*/ */
package akka.stream.tck package akka.stream.tck
import akka.stream.impl.ConstantFun
import akka.stream.scaladsl.Sink import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source import akka.stream.scaladsl.Source
import akka.util.ConstantFun
import org.reactivestreams.Publisher import org.reactivestreams.Publisher
class FlattenTest extends AkkaPublisherVerification[Int] { class FlattenTest extends AkkaPublisherVerification[Int] {

2
akka-stream-tests/src/test/java/akka/stream/javadsl/FlowTest.java
View file

@ -10,7 +10,7 @@ import akka.japi.JavaPartialFunction;
import akka.japi.Pair; import akka.japi.Pair;
import akka.japi.function.*; import akka.japi.function.*;
import akka.stream.*; import akka.stream.*;
import akka.stream.impl.ConstantFun; import akka.util.ConstantFun;
import akka.stream.javadsl.GraphDSL.Builder; import akka.stream.javadsl.GraphDSL.Builder;
import akka.stream.stage.*; import akka.stream.stage.*;
import akka.testkit.AkkaSpec; import akka.testkit.AkkaSpec;

2
akka-stream-tests/src/test/java/akka/stream/javadsl/SourceTest.java
View file

@ -11,7 +11,7 @@ import akka.japi.Pair;
import akka.japi.function.*; import akka.japi.function.*;
import akka.japi.pf.PFBuilder; import akka.japi.pf.PFBuilder;
import akka.stream.*; import akka.stream.*;
import akka.stream.impl.ConstantFun; import akka.util.ConstantFun;
import akka.stream.stage.*; import akka.stream.stage.*;
import akka.testkit.AkkaSpec; import akka.testkit.AkkaSpec;
import akka.stream.testkit.TestPublisher; import akka.stream.testkit.TestPublisher;

2
akka-stream-tests/src/test/scala/akka/stream/impl/fusing/InterpreterSpec.scala
View file

@ -4,12 +4,12 @@
package akka.stream.impl.fusing package akka.stream.impl.fusing
import akka.stream.impl.ConstantFun
import akka.stream.stage._ import akka.stream.stage._
import akka.stream.testkit.StreamSpec import akka.stream.testkit.StreamSpec
import akka.testkit.EventFilter import akka.testkit.EventFilter
import akka.stream._ import akka.stream._
import akka.stream.impl.fusing.GraphStages.SimpleLinearGraphStage import akka.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
import akka.util.ConstantFun
class InterpreterSpec extends StreamSpec with GraphInterpreterSpecKit { class InterpreterSpec extends StreamSpec with GraphInterpreterSpecKit {
import Supervision.stoppingDecider import Supervision.stoppingDecider

2
akka-stream-tests/src/test/scala/akka/stream/impl/fusing/InterpreterStressSpec.scala
View file

@ -4,10 +4,10 @@
package akka.stream.impl.fusing package akka.stream.impl.fusing
import akka.stream.impl.ConstantFun
import akka.stream.Supervision import akka.stream.Supervision
import akka.stream.testkit.StreamSpec import akka.stream.testkit.StreamSpec
import akka.testkit.LongRunningTest import akka.testkit.LongRunningTest
import akka.util.ConstantFun
class InterpreterStressSpec extends StreamSpec with GraphInterpreterSpecKit { class InterpreterStressSpec extends StreamSpec with GraphInterpreterSpecKit {
import Supervision.stoppingDecider import Supervision.stoppingDecider

2
akka-stream-tests/src/test/scala/akka/stream/scaladsl/FlowConcatAllSpec.scala
View file

@ -4,7 +4,7 @@
package akka.stream.scaladsl package akka.stream.scaladsl
import akka.NotUsed import akka.NotUsed
import akka.stream.impl.ConstantFun import akka.util.ConstantFun
import scala.util.control.NoStackTrace import scala.util.control.NoStackTrace
import akka.stream.ActorMaterializer import akka.stream.ActorMaterializer

1
akka-stream/src/main/scala/akka/stream/impl/ConstantFun.scala
View file

@ -10,6 +10,7 @@ import akka.japi.{ Pair ⇒ JPair }
/** /**
* INTERNAL API * INTERNAL API
*/ */
@deprecated("Use akka.util.ConstantFun instead", "2.5.0")
@InternalApi private[akka] object ConstantFun { @InternalApi private[akka] object ConstantFun {
private[this] val JavaIdentityFunction = new JFun[Any, Any] { private[this] val JavaIdentityFunction = new JFun[Any, Any] {
@throws(classOf[Exception]) override def apply(param: Any): Any = param @throws(classOf[Exception]) override def apply(param: Any): Any = param

3
akka-stream/src/main/scala/akka/stream/javadsl/Flow.scala
View file

@ -3,10 +3,11 @@
*/ */
package akka.stream.javadsl package akka.stream.javadsl
import akka.util.ConstantFun
import akka.{ Done, NotUsed } import akka.{ Done, NotUsed }
import akka.event.LoggingAdapter import akka.event.LoggingAdapter
import akka.japi.{ Pair, function } import akka.japi.{ Pair, function }
import akka.stream.impl.{ ConstantFun, StreamLayout } import akka.stream.impl.StreamLayout
import akka.stream._ import akka.stream._
import org.reactivestreams.Processor import org.reactivestreams.Processor

2
akka-stream/src/main/scala/akka/stream/javadsl/Graph.scala
View file

@ -6,8 +6,8 @@ package akka.stream.javadsl
import akka.NotUsed import akka.NotUsed
import akka.stream._ import akka.stream._
import akka.japi.{ Pair, function } import akka.japi.{ Pair, function }
import akka.util.ConstantFun
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance
import akka.stream.impl.ConstantFun
/** /**
* Merge several streams, taking elements as they arrive from input streams * Merge several streams, taking elements as they arrive from input streams

3
akka-stream/src/main/scala/akka/stream/javadsl/Source.scala
View file

@ -6,12 +6,13 @@ package akka.stream.javadsl
import java.util import java.util
import java.util.Optional import java.util.Optional
import akka.util.ConstantFun
import akka.{ Done, NotUsed } import akka.{ Done, NotUsed }
import akka.actor.{ ActorRef, Cancellable, Props } import akka.actor.{ ActorRef, Cancellable, Props }
import akka.event.LoggingAdapter import akka.event.LoggingAdapter
import akka.japi.{ Pair, Util, function } import akka.japi.{ Pair, Util, function }
import akka.stream._ import akka.stream._
import akka.stream.impl.{ ConstantFun, LinearTraversalBuilder, SourceQueueAdapter, StreamLayout } import akka.stream.impl.{ LinearTraversalBuilder, SourceQueueAdapter, StreamLayout }
import org.reactivestreams.{ Publisher, Subscriber } import org.reactivestreams.{ Publisher, Subscriber }
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance

2
akka-stream/src/main/scala/akka/stream/javadsl/SubFlow.scala
View file

@ -7,7 +7,7 @@ import akka.NotUsed
import akka.event.LoggingAdapter import akka.event.LoggingAdapter
import akka.japi.function import akka.japi.function
import akka.stream._ import akka.stream._
import akka.stream.impl.ConstantFun import akka.util.ConstantFun
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance

2
akka-stream/src/main/scala/akka/stream/javadsl/SubSource.scala
View file

@ -7,7 +7,7 @@ import akka.NotUsed
import akka.event.LoggingAdapter import akka.event.LoggingAdapter
import akka.japi.function import akka.japi.function
import akka.stream._ import akka.stream._
import akka.stream.impl.ConstantFun import akka.util.ConstantFun
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance

1
akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala
View file

@ -9,6 +9,7 @@ import akka.Done
import akka.stream.impl._ import akka.stream.impl._
import akka.stream.impl.fusing._ import akka.stream.impl.fusing._
import akka.stream.stage._ import akka.stream.stage._
import akka.util.ConstantFun
import org.reactivestreams.{ Processor, Publisher, Subscriber, Subscription } import org.reactivestreams.{ Processor, Publisher, Subscriber, Subscription }
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance

1
akka-stream/src/main/scala/akka/stream/scaladsl/Graph.scala
View file

@ -10,6 +10,7 @@ import akka.stream.impl.fusing.GraphStages
import akka.stream.impl.Stages.DefaultAttributes import akka.stream.impl.Stages.DefaultAttributes
import akka.stream.scaladsl.Partition.PartitionOutOfBoundsException import akka.stream.scaladsl.Partition.PartitionOutOfBoundsException
import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler } import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler }
import akka.util.ConstantFun
import scala.annotation.unchecked.uncheckedVariance import scala.annotation.unchecked.uncheckedVariance
import scala.annotation.tailrec import scala.annotation.tailrec

1
akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala
View file

@ -4,6 +4,7 @@
package akka.stream.scaladsl package akka.stream.scaladsl
import akka.stream.impl.Stages.DefaultAttributes import akka.stream.impl.Stages.DefaultAttributes
import akka.util.ConstantFun
import akka.{ Done, NotUsed } import akka.{ Done, NotUsed }
import akka.actor.{ ActorRef, Cancellable, Props } import akka.actor.{ ActorRef, Cancellable, Props }
import akka.stream.actor.ActorPublisher import akka.stream.actor.ActorPublisher

58
project/MiMa.scala
View file

@ -136,7 +136,7 @@ object MiMa extends AutoPlugin {
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.CircuitBreaker#State.callThrough"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.CircuitBreaker#State.callThrough"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.CircuitBreaker#State.invoke"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.CircuitBreaker#State.invoke"),
// #21423 Remove deprecated metrics // #21423 Remove deprecated metrics
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterReadView.clusterMetrics"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ClusterReadView.clusterMetrics"),
ProblemFilters.exclude[MissingClassProblem]("akka.cluster.InternalClusterAction$MetricsTick$"), ProblemFilters.exclude[MissingClassProblem]("akka.cluster.InternalClusterAction$MetricsTick$"),
ProblemFilters.exclude[MissingClassProblem]("akka.cluster.MetricsCollector"), ProblemFilters.exclude[MissingClassProblem]("akka.cluster.MetricsCollector"),
@ -866,7 +866,7 @@ object MiMa extends AutoPlugin {
// Interpreter internals change // Interpreter internals change
ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.stream.stage.GraphStageLogic.portToConn"), ProblemFilters.exclude[IncompatibleResultTypeProblem]("akka.stream.stage.GraphStageLogic.portToConn"),
// #20994 adding new decode method, since we're on JDK7+ now // #20994 adding new decode method, since we're on JDK7+ now
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.util.ByteString.decodeString"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.util.ByteString.decodeString"),
// #20508 HTTP: Document how to be able to support custom request methods // #20508 HTTP: Document how to be able to support custom request methods
@ -1167,37 +1167,37 @@ object MiMa extends AutoPlugin {
val Release25Filters = Seq( val Release25Filters = Seq(
"2.5.0" -> Seq( "2.5.0" -> Seq(
// #22759 LMDB files // #22759 LMDB files
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.env"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.env"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.db"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.db"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.keyBuffer"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.keyBuffer"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.valueBuffer_="), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.valueBuffer_="),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.valueBuffer"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.cluster.ddata.LmdbDurableStore.valueBuffer"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.stream.scaladsl.FlowOps.groupedWeightedWithin"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.stream.scaladsl.FlowOps.groupedWeightedWithin"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSubscriber.props"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSubscriber.props"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSource.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSource.this"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSink.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSink.this"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FilePublisher.props"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FilePublisher.props"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSubscriber.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FileSubscriber.this"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FilePublisher.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.impl.io.FilePublisher.this"),
ProblemFilters.exclude[MissingClassProblem]("akka.stream.impl.fusing.GroupedWithin"), ProblemFilters.exclude[MissingClassProblem]("akka.stream.impl.fusing.GroupedWithin"),
ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.traversalBuilder"), ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.traversalBuilder"),
ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.named"), ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.named"),
ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.addAttributes"), ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.addAttributes"),
ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.async") ProblemFilters.exclude[InheritedNewAbstractMethodProblem]("akka.stream.Graph.async")
), ),
"2.5.1" -> Seq( "2.5.1" -> Seq(
// #21213 Feature request: Let BackoffSupervisor reply to messages when its child is stopped // #21213 Feature request: Let BackoffSupervisor reply to messages when its child is stopped
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffSupervisor.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffSupervisor.this"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.copy"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.copy"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.this"),
ProblemFilters.exclude[MissingTypesProblem]("akka.pattern.BackoffOptionsImpl$"), ProblemFilters.exclude[MissingTypesProblem]("akka.pattern.BackoffOptionsImpl$"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.apply"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOptionsImpl.apply"),
ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOnRestartSupervisor.this"), ProblemFilters.exclude[DirectMissingMethodProblem]("akka.pattern.BackoffOnRestartSupervisor.this"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.HandleBackoff.replyWhileStopped"), ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.HandleBackoff.replyWhileStopped"),
ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.BackoffOptions.withReplyWhileStopped") ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.pattern.BackoffOptions.withReplyWhileStopped")
) )
) )