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

align Effect API between scaladsl and javadsl, #25706

Browse source 
scaladsl:
* stop => stop()
* andThenStop() => thenStop()

javadsl:
* andThen => thenRun
This commit is contained in:
Patrik Nordwall 2018-10-19 08:47:34 +02:00
parent 751a67ef42
commit a948f5572b
21 changed files with 79 additions and 54 deletions
Download patch file Download diff file Expand all files Collapse all files

2
akka-cluster-sharding-typed/src/test/java/jdocs/akka/cluster/sharding/typed/HelloWorldPersistentEntityExample.java
View file

@ -150,7 +150,7 @@ public class HelloWorldPersistentEntityExample {
private Effect<Greeted, KnownPeople> greet(KnownPeople state, Greet cmd) {
return Effect().persist(new Greeted(cmd.whom))
.andThen(newState -> cmd.replyTo.tell(new Greeting(cmd.whom, newState.numberOfPeople())));
.thenRun(newState -> cmd.replyTo.tell(new Greeting(cmd.whom, newState.numberOfPeople())));
}
@Override

2
akka-cluster-sharding-typed/src/test/scala/akka/cluster/sharding/typed/scaladsl/ClusterShardingPersistenceSpec.scala
View file

@ -54,7 +54,7 @@ object ClusterShardingPersistenceSpec {
case Get(replyTo) ⇒
replyTo ! s"$entityId:$state"
Effect.none
case StopPlz ⇒ Effect.stop
case StopPlz ⇒ Effect.stop()
},
eventHandler = (state, evt) ⇒ if (state.isEmpty) evt else state + "|" + evt)

2
akka-cluster-sharding-typed/src/test/scala/docs/akka/cluster/sharding/typed/HelloWorldPersistentEntityExample.scala
View file

@ -77,7 +77,7 @@ object HelloWorldPersistentEntityExample {
.thenRun(state ⇒ cmd.replyTo ! Greeting(cmd.whom, state.numberOfPeople))
private def passivate(): Effect[Greeted, KnownPeople] =
Effect.stop
Effect.stop()
private val eventHandler: (KnownPeople, Greeted) ⇒ KnownPeople = {
(state, evt) ⇒ state.add(evt.whom)

2
akka-cluster-typed/src/test/scala/akka/cluster/typed/ClusterSingletonPersistenceSpec.scala
View file

@ -45,7 +45,7 @@ object ClusterSingletonPersistenceSpec {
case Get(replyTo) ⇒
replyTo ! state
Effect.none
case StopPlz ⇒ Effect.stop
case StopPlz ⇒ Effect.stop()
},
eventHandler = (state, evt) ⇒ if (state.isEmpty) evt else state + "|" + evt)

5
akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EffectImpl.scala
View file

@ -7,9 +7,9 @@ package akka.persistence.typed.internal
import scala.collection.immutable
import akka.persistence.typed.{ SideEffect, javadsl, scaladsl }
import akka.annotation.InternalApi
import akka.persistence.typed.NoReplyEffectImpl
import akka.persistence.typed.Stop
/** INTERNAL API */
@InternalApi
@ -23,6 +23,9 @@ private[akka] abstract class EffectImpl[+Event, State] extends javadsl.ReplyEffe
override def thenNoReply(): EffectImpl[Event, State] =
CompositeEffect(this, new NoReplyEffectImpl[State])
override def thenStop(): EffectImpl[Event, State] =
CompositeEffect(this, Stop.asInstanceOf[SideEffect[State]])
}
/** INTERNAL API */

42
akka-persistence-typed/src/main/scala/akka/persistence/typed/javadsl/Effect.scala
View file

@ -7,15 +7,20 @@ package akka.persistence.typed.javadsl
import akka.annotation.DoNotInherit
import akka.japi.function
import akka.persistence.typed.internal._
import akka.persistence.typed.{ SideEffect, Stop }
import akka.persistence.typed.SideEffect
import scala.collection.JavaConverters._
import akka.annotation.InternalApi
import akka.persistence.typed.ExpectingReply
object EffectFactory extends EffectFactories[Nothing, Nothing, Nothing]
/**
* INTERNAL API: see `class EffectFactories`
*/
@InternalApi private[akka] object EffectFactories extends EffectFactories[Nothing, Nothing, Nothing]
/**
* Factory methods for creating [[Effect]] directives.
* Factory methods for creating [[Effect]] directives - how a persistent actor reacts on a command.
* Created via [[PersistentBehavior.Effect]].
*
* Not for user extension
*/
@ -35,17 +40,17 @@ object EffectFactory extends EffectFactories[Nothing, Nothing, Nothing]
/**
* Do not persist anything
*/
def none: Effect[Event, State] = PersistNothing.asInstanceOf[Effect[Event, State]]
def none(): Effect[Event, State] = PersistNothing.asInstanceOf[Effect[Event, State]]
/**
* Stop this persistent actor
*/
def stop: Effect[Event, State] = none.thenStop()
def stop(): Effect[Event, State] = none().thenStop()
/**
* This command is not handled, but it is not an error that it isn't.
*/
def unhandled: Effect[Event, State] = Unhandled.asInstanceOf[Effect[Event, State]]
def unhandled(): Effect[Event, State] = Unhandled.asInstanceOf[Effect[Event, State]]
/**
* Send a reply message to the command, which implements [[ExpectingReply]]. The type of the
@ -60,7 +65,7 @@ object EffectFactory extends EffectFactories[Nothing, Nothing, Nothing]
* finding mistakes.
*/
def reply[ReplyMessage](cmd: ExpectingReply[ReplyMessage], replyWithMessage: ReplyMessage): ReplyEffect[Event, State] =
none.thenReply[ReplyMessage](cmd, new function.Function[State, ReplyMessage] {
none().thenReply[ReplyMessage](cmd, new function.Function[State, ReplyMessage] {
override def apply(param: State): ReplyMessage = replyWithMessage
})
@ -70,7 +75,7 @@ object EffectFactory extends EffectFactories[Nothing, Nothing, Nothing]
* sent for a specific command or the reply will be sent later.
*/
def noReply(): ReplyEffect[Event, State] =
none.thenNoReply()
none().thenNoReply()
}
/**
@ -78,24 +83,31 @@ object EffectFactory extends EffectFactories[Nothing, Nothing, Nothing]
*
* Additional side effects can be performed in the callback `andThen`
*
* Instances of `Effect` are available through factories in the respective Java and Scala DSL packages.
* Instances of `Effect` are available through factories [[PersistentBehavior.Effect]].
*
* Not intended for user extension.
*/
@DoNotInherit abstract class Effect[+Event, State] {
self: EffectImpl[Event, State] ⇒
/** Convenience method to register a side effect with just a callback function */
final def andThen(callback: function.Procedure[State]): Effect[Event, State] =
/**
* Run the given callback. Callbacks are run sequentially.
*/
final def thenRun(callback: function.Procedure[State]): Effect[Event, State] =
CompositeEffect(this, SideEffect[State](s ⇒ callback.apply(s)))
/** Convenience method to register a side effect that doesn't need access to state */
final def andThen(callback: function.Effect): Effect[Event, State] =
/**
* Run the given callback. Callbacks are run sequentially.
*/
final def thenRun(callback: function.Effect): Effect[Event, State] =
CompositeEffect(this, SideEffect[State]((_: State) ⇒ callback.apply()))
/**
* Run the given callback after the current Effect
*/
def andThen(chainedEffect: SideEffect[State]): Effect[Event, State]
final def thenStop(): Effect[Event, State] =
CompositeEffect(this, Stop.asInstanceOf[SideEffect[State]])
/** The side effect is to stop the actor */
def thenStop(): Effect[Event, State]
/**
* Send a reply message to the command, which implements [[ExpectingReply]]. The type of the

13
akka-persistence-typed/src/main/scala/akka/persistence/typed/javadsl/EventHandler.scala
View file

@ -82,6 +82,19 @@ final class EventHandlerBuilder[State >: Null, Event]() {
build()
}
/**
* Match any event.
*
* Use this when then `State` is not needed in the `handler`, otherwise there is an overloaded method that pass
* the state in a `BiFunction`.
*/
def matchAny(f: JFunction[Event, State]): EventHandler[State, Event] = {
matchAny(new BiFunction[State, Event, State] {
override def apply(state: State, event: Event): State = f(event)
})
build()
}
/**
* Compose this builder with another builder. The handlers in this builder will be tried first followed
* by the handlers in `other`.

3
akka-persistence-typed/src/main/scala/akka/persistence/typed/javadsl/PersistentBehavior.scala
View file

@ -32,7 +32,8 @@ abstract class PersistentBehavior[Command, Event, State >: Null] private[akka] (
*
* Return effects from your handlers in order to instruct persistence on how to act on the incoming message (i.e. persist events).
*/
protected final def Effect: EffectFactories[Command, Event, State] = EffectFactory.asInstanceOf[EffectFactories[Command, Event, State]]
protected final def Effect: EffectFactories[Command, Event, State] =
EffectFactories.asInstanceOf[EffectFactories[Command, Event, State]]
/**
* Implement by returning the initial empty state object.

8
akka-persistence-typed/src/main/scala/akka/persistence/typed/scaladsl/Effect.scala
View file

@ -13,7 +13,7 @@ import akka.persistence.typed.ExpectingReply
import akka.persistence.typed.ReplyEffectImpl
/**
* Factories for effects - how a persistent actor reacts on a command
* Factory methods for creating [[Effect]] directives - how a persistent actor reacts on a command.
*/
object Effect {
@ -58,7 +58,7 @@ object Effect {
* Stop this persistent actor
* Side effects can be chained with `andThen`
*/
def stop[Event, State]: Effect[Event, State] = none.andThenStop()
def stop[Event, State](): Effect[Event, State] = none.thenStop()
/**
* Send a reply message to the command, which implements [[ExpectingReply]]. The type of the
@ -113,9 +113,7 @@ trait Effect[+Event, State] {
CompositeEffect(this, chainedEffects)
/** The side effect is to stop the actor */
def andThenStop(): Effect[Event, State] = {
CompositeEffect(this, Stop.asInstanceOf[SideEffect[State]])
}
def thenStop(): Effect[Event, State]
/**
* Send a reply message to the command, which implements [[ExpectingReply]]. The type of the

4
akka-persistence-typed/src/test/java/akka/persistence/typed/javadsl/PersistentActorCompileOnlyTest.java
View file

@ -176,7 +176,7 @@ public class PersistentActorCompileOnlyTest {
//#commonChainedEffects
return commandHandlerBuilder(ExampleState.class)
.matchCommand(Cmd.class, (state, cmd) -> Effect().persist(new Evt(cmd.data))
.andThen(() -> cmd.sender.tell(new Ack()))
.thenRun(() -> cmd.sender.tell(new Ack()))
.andThen(commonChainedEffect)
)
.build();
@ -303,7 +303,7 @@ public class PersistentActorCompileOnlyTest {
return commandHandlerBuilder(EventsInFlight.class)
.matchCommand(DoSideEffect.class,
(state, cmd) -> Effect().persist(new IntentRecord(state.nextCorrelationId, cmd.data))
.andThen(() -> performSideEffect(ctx.getSelf().narrow(), state.nextCorrelationId, cmd.data, ctx.getSystem().scheduler())))
.thenRun(() -> performSideEffect(ctx.getSelf().narrow(), state.nextCorrelationId, cmd.data, ctx.getSystem().scheduler())))
.matchCommand(AcknowledgeSideEffect.class, (state, command) -> Effect().persist(new SideEffectAcknowledged(command.correlationId)))
.build();
}

6
akka-persistence-typed/src/test/java/akka/persistence/typed/javadsl/PersistentActorJavaDslTest.java
View file

@ -292,14 +292,14 @@ public class PersistentActorJavaDslTest extends JUnitSuite {
.matchCommand(Timeout.class,
(state, msg) -> Effect().persist(timeoutEvent))
.matchCommand(EmptyEventsListAndThenLog.class, (state, msg) -> Effect().persist(Collections.emptyList())
.andThen(s -> loggingProbe.tell(loggingOne)))
.thenRun(s -> loggingProbe.tell(loggingOne)))
.matchCommand(StopThenLog.class,
(state, msg) -> Effect().stop()
.andThen(s -> loggingProbe.tell(loggingOne)))
.thenRun(s -> loggingProbe.tell(loggingOne)))
.matchCommand(IncrementTwiceAndLog.class,
(state, msg) -> Effect().persist(
Arrays.asList(new Incremented(1), new Incremented(1)))
.andThen(s -> loggingProbe.tell(loggingOne)))
.thenRun(s -> loggingProbe.tell(loggingOne)))
.build();
}

2
akka-persistence-typed/src/test/java/jdocs/akka/persistence/typed/AccountExample.java
View file

@ -88,7 +88,7 @@ public class AccountExample extends PersistentBehavior<AccountExample.AccountCom
return Effect().unhandled(); // TODO replies are missing in this example
} else {
return Effect().persist(new Withdrawn(cmd.amount))
.andThen(acc2 -> { // FIXME in scaladsl it's named thenRun, change javadsl also?
.thenRun(acc2 -> { // FIXME in scaladsl it's named thenRun, change javadsl also?
// we know this cast is safe, but somewhat ugly
OpenedAccount openAccount = (OpenedAccount) acc2;
// do some side-effect using balance

10
akka-persistence-typed/src/test/java/jdocs/akka/persistence/typed/BlogPostExample.java
View file

@ -15,8 +15,6 @@ import akka.persistence.typed.javadsl.CommandHandlerBuilder;
import akka.persistence.typed.javadsl.EventHandler;
import akka.persistence.typed.javadsl.PersistentBehavior;
import java.util.Optional;
public class BlogPostExample {
//#event
@ -168,7 +166,7 @@ public class BlogPostExample {
//#reply
PostAdded event = new PostAdded(cmd.content.postId, cmd.content);
return Effect().persist(event)
.andThen(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
.thenRun(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
//#reply
});
}
@ -179,10 +177,10 @@ public class BlogPostExample {
return commandHandlerBuilder(DraftState.class)
.matchCommand(ChangeBody.class, (state, cmd) -> {
BodyChanged event = new BodyChanged(state.postId(), cmd.newBody);
return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
return Effect().persist(event).thenRun(() -> cmd.replyTo.tell(Done.getInstance()));
})
.matchCommand(Publish.class, (state, cmd) -> Effect()
.persist(new Published(state.postId())).andThen(() -> {
.persist(new Published(state.postId())).thenRun(() -> {
System.out.println("Blog post published: " + state.postId());
cmd.replyTo.tell(Done.getInstance());
}))
@ -196,7 +194,7 @@ public class BlogPostExample {
return commandHandlerBuilder(PublishedState.class)
.matchCommand(ChangeBody.class, (state, cmd) -> {
BodyChanged event = new BodyChanged(state.postId(), cmd.newBody);
return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
return Effect().persist(event).thenRun(() -> cmd.replyTo.tell(Done.getInstance()));
})
.matchCommand(GetPost.class, (state, cmd) -> {
cmd.replyTo.tell(state.postContent);

6
akka-persistence-typed/src/test/java/jdocs/akka/persistence/typed/NullBlogState.java
View file

@ -127,7 +127,7 @@ public class NullBlogState {
.matchCommand(AddPost.class, cmd -> {
PostAdded event = new PostAdded(cmd.content.postId, cmd.content);
return Effect().persist(event)
.andThen(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
.thenRun(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
})
.matchCommand(PassivatePost.class, cmd -> Effect().stop());
}
@ -136,10 +136,10 @@ public class NullBlogState {
return commandHandlerBuilder(Objects::nonNull)
.matchCommand(ChangeBody.class, (state, cmd) -> {
BodyChanged event = new BodyChanged(state.postId(), cmd.newBody);
return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
return Effect().persist(event).thenRun(() -> cmd.replyTo.tell(Done.getInstance()));
})
.matchCommand(Publish.class, (state, cmd) -> Effect()
.persist(new Published(state.postId())).andThen(() -> {
.persist(new Published(state.postId())).thenRun(() -> {
System.out.println("Blog post published: " + state.postId());
cmd.replyTo.tell(Done.getInstance());
}))

6
akka-persistence-typed/src/test/java/jdocs/akka/persistence/typed/OptionalBlogState.java
View file

@ -127,7 +127,7 @@ public class OptionalBlogState {
.matchCommand(AddPost.class, (state, cmd) -> {
PostAdded event = new PostAdded(cmd.content.postId, cmd.content);
return Effect().persist(event)
.andThen(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
.thenRun(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
})
.matchCommand(PassivatePost.class, cmd -> Effect().stop());
}
@ -136,10 +136,10 @@ public class OptionalBlogState {
return commandHandlerBuilder(state -> state.isPresent())
.matchCommand(ChangeBody.class, (state, cmd) -> {
BodyChanged event = new BodyChanged(state.get().postId(), cmd.newBody);
return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
return Effect().persist(event).thenRun(() -> cmd.replyTo.tell(Done.getInstance()));
})
.matchCommand(Publish.class, (state, cmd) -> Effect()
.persist(new Published(state.get().postId())).andThen(() -> {
.persist(new Published(state.get().postId())).thenRun(() -> {
System.out.println("Blog post published: " + state.get().postId());
cmd.replyTo.tell(Done.getInstance());
}))

2
akka-persistence-typed/src/test/scala/akka/persistence/typed/internal/RecoveryPermitterSpec.scala
View file

@ -48,7 +48,7 @@ object RecoveryPermitterSpec {
persistenceId = PersistenceId(name),
emptyState = EmptyState,
commandHandler = CommandHandler.command {
case StopActor ⇒ Effect.stop
case StopActor ⇒ Effect.stop()
case command ⇒ commandProbe.ref ! command; Effect.none
},
eventHandler = { (state, event) ⇒ eventProbe.ref ! event; state }

2
akka-persistence-typed/src/test/scala/akka/persistence/typed/scaladsl/NullEmptyStateSpec.scala
View file

@ -30,7 +30,7 @@ class NullEmptyStateSpec extends ScalaTestWithActorTestKit(NullEmptyStateSpec.co
emptyState = null,
commandHandler = (_, command) ⇒ {
if (command == "stop")
Effect.stop
Effect.stop()
else
Effect.persist(command)
},

2
akka-persistence-typed/src/test/scala/akka/persistence/typed/scaladsl/PersistentActorCompileOnlyTest.scala
View file

@ -398,7 +398,7 @@ object PersistentActorCompileOnlyTest {
case Enough ⇒
Effect.persist(Done)
.thenRun((_: State) ⇒ println("yay"))
.andThenStop
.thenStop
}
private val eventHandler: (State, Event) ⇒ State = {

6
akka-persistence-typed/src/test/scala/akka/persistence/typed/scaladsl/PersistentBehaviorSpec.scala
View file

@ -139,14 +139,14 @@ object PersistentBehaviorSpec {
.thenRun { (_: State) ⇒
loggingActor ! firstLogging
}
.andThenStop
.thenStop
case IncrementTwiceThenLogThenStop ⇒
Effect.persist(Incremented(1), Incremented(2))
.thenRun { (_: State) ⇒
loggingActor ! firstLogging
}
.andThenStop
.thenStop
case IncrementWithPersistAll(n) ⇒
Effect.persist((0 until n).map(_ ⇒ Incremented(1)))
@ -210,7 +210,7 @@ object PersistentBehaviorSpec {
.thenRun { _ ⇒
loggingActor ! firstLogging
}
.andThenStop
.thenStop
},
eventHandler = (state, evt) ⇒ evt match {
case Incremented(delta) ⇒

2
akka-persistence-typed/src/test/scala/akka/persistence/typed/scaladsl/PrimitiveStateSpec.scala
View file

@ -30,7 +30,7 @@ class PrimitiveStateSpec extends ScalaTestWithActorTestKit(PrimitiveStateSpec.co
emptyState = 0,
commandHandler = (_, command) ⇒ {
if (command < 0)
Effect.stop
Effect.stop()
else
Effect.persist(command)
},

6
akka-persistence-typed/src/test/scala/docs/akka/persistence/typed/BlogPostExample.scala
View file

@ -70,7 +70,7 @@ object BlogPostExample {
case BlankState ⇒ command match {
case cmd: AddPost ⇒ addPost(cmd)
case PassivatePost ⇒ Effect.stop
case PassivatePost ⇒ Effect.stop()
case _ ⇒ Effect.unhandled
}
@ -79,12 +79,12 @@ object BlogPostExample {
case Publish(replyTo) ⇒ publish(draftState, replyTo)
case GetPost(replyTo) ⇒ getPost(draftState, replyTo)
case _: AddPost ⇒ Effect.unhandled
case PassivatePost ⇒ Effect.stop
case PassivatePost ⇒ Effect.stop()
}
case publishedState: PublishedState ⇒ command match {
case GetPost(replyTo) ⇒ getPost(publishedState, replyTo)
case PassivatePost ⇒ Effect.stop
case PassivatePost ⇒ Effect.stop()
case _ ⇒ Effect.unhandled
}
}