From 24c7109524375939ba7d1409054a8d294d06ea23 Mon Sep 17 00:00:00 2001 From: Artur Souza Date: Mon, 13 Jan 2020 17:47:39 -0800 Subject: [PATCH] Unit tests for stateful actors + fixes. --- .../actors/client/ActorMethodEnvelope.java | 35 - .../io/dapr/actors/client/ActorProxyImpl.java | 6 +- .../io/dapr/actors/runtime/AbstractActor.java | 570 +++++++-------- .../io/dapr/actors/runtime/ActorManager.java | 475 +++++++------ .../actors/runtime/ActorStateManager.java | 526 +++++++------- .../actors/runtime/ActorStateSerializer.java | 19 +- .../dapr/actors/runtime/ActorTimerParams.java | 89 --- .../actors/client/ActorProxyForTestsImpl.java | 17 + .../dapr/actors/runtime/ActorManagerTest.java | 19 +- .../actors/runtime/ActorStatefulTest.java | 660 ++++++++++++++++++ .../runtime/DaprInMemoryStateProvider.java | 78 +++ 11 files changed, 1585 insertions(+), 909 deletions(-) delete mode 100644 sdk-actors/src/main/java/io/dapr/actors/client/ActorMethodEnvelope.java delete mode 100644 sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTimerParams.java create mode 100644 sdk-actors/src/test/java/io/dapr/actors/client/ActorProxyForTestsImpl.java create mode 100644 sdk-actors/src/test/java/io/dapr/actors/runtime/ActorStatefulTest.java create mode 100644 sdk-actors/src/test/java/io/dapr/actors/runtime/DaprInMemoryStateProvider.java diff --git a/sdk-actors/src/main/java/io/dapr/actors/client/ActorMethodEnvelope.java b/sdk-actors/src/main/java/io/dapr/actors/client/ActorMethodEnvelope.java deleted file mode 100644 index 693708a22b..0000000000 --- a/sdk-actors/src/main/java/io/dapr/actors/client/ActorMethodEnvelope.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * Copyright (c) Microsoft Corporation. - * Licensed under the MIT License. - */ - -package io.dapr.actors.client; - -/** - * Request and Response object used to talk to Actors. - */ -public class ActorMethodEnvelope { - - /** - * Data serialized for input/output of Actor methods. - */ - private byte[] data; - - /** - * Gets the data serialized for input/output of Actor methods. - * - * @return Data serialized for input/output of Actor methods. - */ - public byte[] getData() { - return data; - } - - /** - * Sets the data serialized for input/output of Actor methods. - * - * @param data Data serialized for input/output of Actor methods. - */ - public void setData(byte[] data) { - this.data = data; - } -} diff --git a/sdk-actors/src/main/java/io/dapr/actors/client/ActorProxyImpl.java b/sdk-actors/src/main/java/io/dapr/actors/client/ActorProxyImpl.java index 3309f21570..c63cec48c1 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/client/ActorProxyImpl.java +++ b/sdk-actors/src/main/java/io/dapr/actors/client/ActorProxyImpl.java @@ -33,7 +33,7 @@ class ActorProxyImpl implements ActorProxy { private final DaprClient daprClient; /** - * Creates a new instance of {@link ActorProxyAsyncClient}. + * Creates a new instance of {@link ActorProxyImpl}. * * @param actorType actor implementation type of the actor associated with the proxy object. * @param actorId The actorId associated with the proxy @@ -127,7 +127,7 @@ public Mono invokeActorMethod(String methodName, Object data) { * @return Response object, null or RuntimeException. */ private T unwrap(final String response, Class clazz) { - return this.serializer.unwrapMethodResponse(response, clazz); + return this.serializer.unwrapData(response, clazz); } /** @@ -139,7 +139,7 @@ private T unwrap(final String response, Class clazz) { * @throws IOException In case it cannot generate String. */ private String wrap(final T request) throws IOException { - return this.serializer.wrapMethodRequest(request); + return this.serializer.wrapData(request); } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java index 4e894fcc40..b670a5a8b2 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/AbstractActor.java @@ -23,286 +23,296 @@ */ public abstract class AbstractActor { - /** - * Type of tracing messages. - */ - private static final String TRACE_TYPE = "Actor"; - - /** - * Context for the Actor runtime. - */ - private final ActorRuntimeContext actorRuntimeContext; - - /** - * Actor identifier. - */ - private final ActorId id; - - /** - * Emits trace messages for Actors. - */ - private final ActorTrace actorTrace; - - /** - * Registered timers for this Actor. - */ - private final Map timers; - - /** - * Manager for the states in Actors. - */ - protected final ActorStateManager actorStateManager; - - /** - * Instantiates a new Actor. - * - * @param runtimeContext Context for the runtime. - * @param id Actor identifier. - */ - protected AbstractActor(ActorRuntimeContext runtimeContext, ActorId id) { - this.actorRuntimeContext = runtimeContext; - this.id = id; - this.actorStateManager = new ActorStateManager( - runtimeContext.getStateProvider(), - runtimeContext.getActorTypeInformation().getName(), - id); - this.actorTrace = runtimeContext.getActorTrace(); - this.timers = Collections.synchronizedMap(new HashMap<>()); - } - - /** - * Returns the id of the actor. - * - * @return Actor id. - */ - protected ActorId getId() { - return this.id; - } - - /** - * Registers a reminder for this Actor. - * - * @param reminderName Name of the reminder. - * @param state State to be send along with reminder triggers. - * @param dueTime Due time for the first trigger. - * @param period Frequency for the triggers. - * @param Type of the state object. - * @return Asynchronous void response. - */ - protected Mono registerReminder( - String reminderName, - T state, - Duration dueTime, - Duration period) { - try { - String data = this.actorRuntimeContext.getActorSerializer().serializeString(state); - ActorReminderParams params = new ActorReminderParams(data, dueTime, period); - String serialized = this.actorRuntimeContext.getActorSerializer().serializeString(params); - return this.actorRuntimeContext.getDaprClient().registerActorReminder( - this.actorRuntimeContext.getActorTypeInformation().getName(), - this.id.toString(), - reminderName, - serialized); - } catch (IOException e) { - return Mono.error(e); - } - } - - /** - * Registers a Timer for the actor. A timer name is autogenerated by the runtime to keep track of it. - * - * @param timerName Name of the timer, unique per Actor (auto-generated if null). - * @param callback Name of the method to be called. - * @param state State to be passed it to the method when timer triggers. - * @param dueTime The amount of time to delay before the async callback is first invoked. - * Specify negative one (-1) milliseconds to prevent the timer from starting. - * Specify zero (0) to start the timer immediately. - * @param period The time interval between invocations of the async callback. - * Specify negative one (-1) milliseconds to disable periodic signaling. - * @param Type for the state to be passed in to timer. - * @return Asynchronous result. - */ - protected Mono registerActorTimer( - String timerName, - String callback, - T state, - Duration dueTime, - Duration period) { - if ((callback == null) || callback.isEmpty()) { - throw new IllegalArgumentException("Timer requires a callback function."); - } - - String name = timerName; - if ((timerName == null) || (timerName.isEmpty())) { - name = String.format("%s_Timer_%d", this.id.toString(), this.timers.size() + 1); - } - - try { - ActorTimer actorTimer = new ActorTimer(this, name, callback, state, dueTime, period); - String serializedTimer = this.actorRuntimeContext.getActorSerializer().serializeString(actorTimer); - - this.timers.put(name, actorTimer); - return this.actorRuntimeContext.getDaprClient().registerActorTimer( - this.actorRuntimeContext.getActorTypeInformation().getName(), - this.id.toString(), - name, - serializedTimer); - } catch (IOException e) { - return Mono.error(e); - } - } - - /** - * Unregisters an Actor timer. - * - * @param actorTimer Timer to be unregistered. - * @return Asynchronous void response. - */ - protected Mono unregister(ActorTimer actorTimer) { - return this.actorRuntimeContext.getDaprClient().unregisterActorTimer( - this.actorRuntimeContext.getActorTypeInformation().getName(), - this.id.toString(), - actorTimer.getName()) - .then(this.onUnregisteredTimer(actorTimer)); - } - - /** - * Callback function invoked after an Actor has been activated. - * - * @return Asynchronous void response. - */ - protected Mono onActivate() { - return Mono.empty(); - } - - /** - * Callback function invoked after an Actor has been deactivated. - * - * @return Asynchronous void response. - */ - protected Mono onDeactivate() { - return Mono.empty(); - } - - /** - * Callback function invoked before method is invoked. - * - * @param actorMethodContext Method context. - * @return Asynchronous void response. - */ - protected Mono onPreActorMethod(ActorMethodContext actorMethodContext) { - return Mono.empty(); - } - - /** - * Callback function invoked after method is invoked. - * - * @param actorMethodContext Method context. - * @return Asynchronous void response. - */ - protected Mono onPostActorMethod(ActorMethodContext actorMethodContext) { - return Mono.empty(); - } - - /** - * Saves the state of this Actor. - * - * @return Asynchronous void response. - */ - protected Mono saveState() { - return this.actorStateManager.save(); - } - - /** - * Resets the state of this Actor. - * - * @return Asynchronous void response. - */ - Mono resetState() { - return this.actorStateManager.clear(); - } - - /** - * Gets a given timer by name. - * - * @param timerName Timer name. - * @return Asynchronous void response. - */ - ActorTimer getActorTimer(String timerName) { - return timers.getOrDefault(timerName, null); - } - - /** - * Internal callback when an Actor is activated. - * - * @return Asynchronous void response. - */ - Mono onActivateInternal() { - this.actorTrace.writeInfo(TRACE_TYPE, this.id.toString(), "Activating ..."); - - return this.resetState() - .then(this.onActivate()) - .then(this.doWriteInfo(TRACE_TYPE, this.id.toString(), "Activated")) - .then(this.saveState()); - } - - /** - * Internal callback when an Actor is deactivated. - * - * @return Asynchronous void response. - */ - Mono onDeactivateInternal() { - this.actorTrace.writeInfo(TRACE_TYPE, this.id.toString(), "Deactivating ..."); - - return this.resetState() - .then(this.onDeactivate()) - .then(this.doWriteInfo(TRACE_TYPE, this.id.toString(), "Deactivated")) - .then(this.saveState()); - } - - /** - * Internal callback prior to method be invoked. - * - * @param actorMethodContext Method context. - * @return Asynchronous void response. - */ - Mono onPreActorMethodInternal(ActorMethodContext actorMethodContext) { - return this.onPreActorMethod(actorMethodContext); - } - - /** - * Internal callback after method is invoked. - * - * @param actorMethodContext Method context. - * @return Asynchronous void response. - */ - Mono onPostActorMethodInternal(ActorMethodContext actorMethodContext) { - return this.onPostActorMethod(actorMethodContext) - .then(this.saveState()); - } - - /** - * Internal callback for when Actor timer is unregistered. - * - * @param timer Timer being unregistered. - * @return Asynchronous void response. - */ - Mono onUnregisteredTimer(ActorTimer timer) { - this.timers.remove(timer.getName()); - return Mono.empty(); - } - - /** - * Internal method to emit a trace message. - * - * @param type Type of trace message. - * @param id Identifier of entity relevant for the trace message. - * @param message Message to be logged. - * @return Asynchronous void response. - */ - private Mono doWriteInfo(String type, String id, String message) { - this.actorTrace.writeInfo(type, id, message); - return Mono.empty(); + /** + * Type of tracing messages. + */ + private static final String TRACE_TYPE = "Actor"; + + /** + * Context for the Actor runtime. + */ + private final ActorRuntimeContext actorRuntimeContext; + + /** + * Actor identifier. + */ + private final ActorId id; + + /** + * Emits trace messages for Actors. + */ + private final ActorTrace actorTrace; + + /** + * Registered timers for this Actor. + */ + private final Map timers; + + /** + * Manager for the states in Actors. + */ + private final ActorStateManager actorStateManager; + + /** + * Instantiates a new Actor. + * + * @param runtimeContext Context for the runtime. + * @param id Actor identifier. + */ + protected AbstractActor(ActorRuntimeContext runtimeContext, ActorId id) { + this.actorRuntimeContext = runtimeContext; + this.id = id; + this.actorStateManager = new ActorStateManager( + runtimeContext.getStateProvider(), + runtimeContext.getActorTypeInformation().getName(), + id); + this.actorTrace = runtimeContext.getActorTrace(); + this.timers = Collections.synchronizedMap(new HashMap<>()); + } + + /** + * Returns the id of the actor. + * + * @return Actor id. + */ + protected ActorId getId() { + return this.id; + } + + /** + * Returns the state store manager for this Actor. + * + * @return State store manager for this Actor + */ + protected ActorStateManager getActorStateManager() { + return this.actorStateManager; + } + + /** + * Registers a reminder for this Actor. + * + * @param reminderName Name of the reminder. + * @param state State to be send along with reminder triggers. + * @param dueTime Due time for the first trigger. + * @param period Frequency for the triggers. + * @param Type of the state object. + * @return Asynchronous void response. + */ + protected Mono registerReminder( + String reminderName, + T state, + Duration dueTime, + Duration period) { + try { + String data = this.actorRuntimeContext.getActorSerializer().serializeString(state); + ActorReminderParams params = new ActorReminderParams(data, dueTime, period); + String serialized = this.actorRuntimeContext.getActorSerializer().serializeString(params); + return this.actorRuntimeContext.getDaprClient().registerActorReminder( + this.actorRuntimeContext.getActorTypeInformation().getName(), + this.id.toString(), + reminderName, + serialized); + } catch (IOException e) { + return Mono.error(e); } + } + + /** + * Registers a Timer for the actor. A timer name is autogenerated by the runtime to keep track of it. + * + * @param timerName Name of the timer, unique per Actor (auto-generated if null). + * @param callback Name of the method to be called. + * @param state State to be passed it to the method when timer triggers. + * @param dueTime The amount of time to delay before the async callback is first invoked. + * Specify negative one (-1) milliseconds to prevent the timer from starting. + * Specify zero (0) to start the timer immediately. + * @param period The time interval between invocations of the async callback. + * Specify negative one (-1) milliseconds to disable periodic signaling. + * @param Type for the state to be passed in to timer. + * @return Asynchronous result. + */ + protected Mono registerActorTimer( + String timerName, + String callback, + T state, + Duration dueTime, + Duration period) { + return Mono.fromSupplier(() -> { + if ((callback == null) || callback.isEmpty()) { + throw new IllegalArgumentException("Timer requires a callback function."); + } + + String name = timerName; + if ((timerName == null) || (timerName.isEmpty())) { + name = String.format("%s_Timer_%d", this.id.toString(), this.timers.size() + 1); + } + + ActorTimer actorTimer = new ActorTimer(this, name, callback, state, dueTime, period); + this.timers.put(name, actorTimer); + return actorTimer; + }).flatMap(actorTimer -> { + try { + return this.actorRuntimeContext.getDaprClient().registerActorTimer( + this.actorRuntimeContext.getActorTypeInformation().getName(), + this.id.toString(), + actorTimer.getName(), + this.actorRuntimeContext.getActorSerializer().serializeString(actorTimer)); + } catch (Exception e) { + return Mono.error(e); + } + }); + } + + /** + * Unregisters an Actor timer. + * + * @param timerName Name of Timer to be unregistered. + * @return Asynchronous void response. + */ + protected Mono unregisterTimer(String timerName) { + return Mono.fromSupplier(() -> getActorTimer(timerName)) + .flatMap(actorTimer -> this.actorRuntimeContext.getDaprClient().unregisterActorTimer( + this.actorRuntimeContext.getActorTypeInformation().getName(), + this.id.toString(), + timerName)) + .then(Mono.fromRunnable(() -> this.timers.remove(timerName))); + } + + /** + * Unregisters a Reminder. + * + * @param reminderName Name of Reminder to be unregistered. + * @return Asynchronous void response. + */ + protected Mono unregisterReminder(String reminderName) { + return this.actorRuntimeContext.getDaprClient().unregisterActorReminder( + this.actorRuntimeContext.getActorTypeInformation().getName(), + this.id.toString(), + reminderName); + } + + /** + * Callback function invoked after an Actor has been activated. + * + * @return Asynchronous void response. + */ + protected Mono onActivate() { + return Mono.empty(); + } + + /** + * Callback function invoked after an Actor has been deactivated. + * + * @return Asynchronous void response. + */ + protected Mono onDeactivate() { + return Mono.empty(); + } + + /** + * Callback function invoked before method is invoked. + * + * @param actorMethodContext Method context. + * @return Asynchronous void response. + */ + protected Mono onPreActorMethod(ActorMethodContext actorMethodContext) { + return Mono.empty(); + } + + /** + * Callback function invoked after method is invoked. + * + * @param actorMethodContext Method context. + * @return Asynchronous void response. + */ + protected Mono onPostActorMethod(ActorMethodContext actorMethodContext) { + return Mono.empty(); + } + + /** + * Saves the state of this Actor. + * + * @return Asynchronous void response. + */ + protected Mono saveState() { + return this.actorStateManager.save(); + } + + /** + * Resets the cached state of this Actor. + */ + void resetState() { + this.actorStateManager.clear(); + } + + /** + * Gets a given timer by name. + * + * @param timerName Timer name. + * @return Asynchronous void response. + */ + ActorTimer getActorTimer(String timerName) { + return timers.getOrDefault(timerName, null); + } + + /** + * Internal callback when an Actor is activated. + * + * @return Asynchronous void response. + */ + Mono onActivateInternal() { + return Mono.fromRunnable(() -> { + this.actorTrace.writeInfo(TRACE_TYPE, this.id.toString(), "Activating ..."); + this.resetState(); + }).then(this.onActivate()) + .then(this.doWriteInfo(TRACE_TYPE, this.id.toString(), "Activated")) + .then(this.saveState()); + } + + /** + * Internal callback when an Actor is deactivated. + * + * @return Asynchronous void response. + */ + Mono onDeactivateInternal() { + this.actorTrace.writeInfo(TRACE_TYPE, this.id.toString(), "Deactivating ..."); + + return Mono.fromRunnable(() -> this.resetState()) + .then(this.onDeactivate()) + .then(this.doWriteInfo(TRACE_TYPE, this.id.toString(), "Deactivated")) + .then(this.saveState()); + } + + /** + * Internal callback prior to method be invoked. + * + * @param actorMethodContext Method context. + * @return Asynchronous void response. + */ + Mono onPreActorMethodInternal(ActorMethodContext actorMethodContext) { + return this.onPreActorMethod(actorMethodContext); + } + + /** + * Internal callback after method is invoked. + * + * @param actorMethodContext Method context. + * @return Asynchronous void response. + */ + Mono onPostActorMethodInternal(ActorMethodContext actorMethodContext) { + return this.onPostActorMethod(actorMethodContext).then(this.saveState()); + } + + /** + * Internal method to emit a trace message. + * + * @param type Type of trace message. + * @param id Identifier of entity relevant for the trace message. + * @param message Message to be logged. + * @return Asynchronous void response. + */ + private Mono doWriteInfo(String type, String id, String message) { + return Mono.fromRunnable(() -> this.actorTrace.writeInfo(type, id, message)); + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java index b915a5d0cc..eda8c2473f 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorManager.java @@ -2,8 +2,10 @@ import io.dapr.actors.ActorId; import reactor.core.publisher.Mono; +import reactor.core.publisher.SignalType; import java.io.IOException; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.util.Collections; import java.util.HashMap; @@ -15,256 +17,289 @@ */ class ActorManager { - /** - * Context for the Actor runtime. - */ - private final ActorRuntimeContext runtimeContext; + /** + * Context for the Actor runtime. + */ + private final ActorRuntimeContext runtimeContext; - /** - * Methods found in Actors. - */ - private final ActorMethodInfoMap actorMethods; + /** + * Methods found in Actors. + */ + private final ActorMethodInfoMap actorMethods; - /** - * Active Actor instances. - */ - private final Map activeActors; + /** + * Active Actor instances. + */ + private final Map activeActors; - /** - * Instantiates a new manager for a given actor referenced in the runtimeContext. - * - * @param runtimeContext Runtime context for the Actor. - */ - ActorManager(ActorRuntimeContext runtimeContext) { - this.runtimeContext = runtimeContext; - this.actorMethods = new ActorMethodInfoMap(runtimeContext.getActorTypeInformation().getInterfaces()); - this.activeActors = Collections.synchronizedMap(new HashMap<>()); - } + /** + * Instantiates a new manager for a given actor referenced in the runtimeContext. + * + * @param runtimeContext Runtime context for the Actor. + */ + ActorManager(ActorRuntimeContext runtimeContext) { + this.runtimeContext = runtimeContext; + this.actorMethods = new ActorMethodInfoMap(runtimeContext.getActorTypeInformation().getInterfaces()); + this.activeActors = Collections.synchronizedMap(new HashMap<>()); + } + + /** + * Activates an Actor. + * + * @param actorId Actor identifier. + * @return Asynchronous void response. + */ + Mono activateActor(ActorId actorId) { + T actor = this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId); - /** - * Activates an Actor. - * - * @param actorId Actor identifier. - * @return Asynchronous void response. - */ - Mono activateActor(ActorId actorId) { - T actor = this.runtimeContext.getActorFactory().createActor(runtimeContext, actorId); + return actor.onActivateInternal().then(this.onActivatedActor(actorId, actor)); + } - return actor.onActivateInternal().then(this.onActivatedActor(actorId, actor)); + /** + * Deactivates an Actor. + * + * @param actorId Actor identifier. + * @return Asynchronous void response. + */ + Mono deactivateActor(ActorId actorId) { + T actor = this.activeActors.remove(actorId); + if (actor != null) { + return actor.onDeactivateInternal(); } - /** - * Deactivates an Actor. - * - * @param actorId Actor identifier. - * @return Asynchronous void response. - */ - Mono deactivateActor(ActorId actorId) { - T actor = this.activeActors.remove(actorId); - if (actor != null) { - return actor.onDeactivateInternal(); - } + return Mono.empty(); + } - return Mono.empty(); + /** + * Invokes a given method in the Actor. + * + * @param actorId Identifier for Actor being invoked. + * @param methodName Name of method being invoked. + * @param request Input object for the method being invoked. + * @return Asynchronous void response. + */ + Mono invokeMethod(ActorId actorId, String methodName, String request) { + return invokeMethod(actorId, null, methodName, request); + } + + /** + * Invokes reminder for Actor. + * + * @param actorId Identifier for Actor being invoked. + * @param reminderName Name of reminder being invoked. + * @param params Parameters for the reminder. + * @return Asynchronous void response. + */ + Mono invokeReminder(ActorId actorId, String reminderName, String params) { + if (!this.runtimeContext.getActorTypeInformation().isRemindable()) { + return Mono.empty(); } - /** - * Invokes a given method in the Actor. - * - * @param actorId Identifier for Actor being invoked. - * @param methodName Name of method being invoked. - * @param request Input object for the method being invoked. - * @return Asynchronous void response. - */ - Mono invokeMethod(ActorId actorId, String methodName, String request) { - return invokeMethod(actorId, null, methodName, request); + try { + ActorReminderParams paramsObject = this + .runtimeContext + .getActorSerializer() + .deserialize(params, ActorReminderParams.class); + return invoke( + actorId, + ActorMethodContext.CreateForReminder(reminderName), + actor -> doReminderInvokation((Remindable) actor, reminderName, paramsObject)) + .then(); + } catch (Exception e) { + return Mono.error(e); } + } - /** - * Invokes reminder for Actor. - * - * @param actorId Identifier for Actor being invoked. - * @param reminderName Name of reminder being invoked. - * @param request Parameters for the reminder. - * @return Asynchronous void response. - */ - Mono invokeReminder(ActorId actorId, String reminderName, String request) { - if (!this.runtimeContext.getActorTypeInformation().isRemindable()) { - return Mono.empty(); - } + /** + * Invokes a timer for a given Actor. + * + * @param actorId Identifier for Actor. + * @param timerName Name of timer being invoked. + * @return Asynchronous void response. + */ + Mono invokeTimer(ActorId actorId, String timerName) { + return Mono.fromSupplier(() -> { + AbstractActor actor = this.activeActors.getOrDefault(actorId, null); + if (actor == null) { + throw new IllegalArgumentException( + String.format("Could not find actor %s of type %s.", + actorId.toString(), + this.runtimeContext.getActorTypeInformation().getName())); + } - try { - ActorReminderParams params = this.runtimeContext.getActorSerializer().deserialize(request, ActorReminderParams.class); + ActorTimer actorTimer = actor.getActorTimer(timerName); + if (actorTimer == null) { + throw new IllegalStateException( + String.format("Could not find timer %s for actor %s.", + timerName, + this.runtimeContext.getActorTypeInformation().getName())); + } - return invoke( - actorId, - ActorMethodContext.CreateForReminder(reminderName), - actor -> doReminderInvokation((Remindable) actor, reminderName, params)) - .then(); - } catch (Exception e) { - return Mono.error(e); - } - } + return actorTimer; + }).flatMap(actorTimer -> invokeMethod( + actorId, + ActorMethodContext.CreateForTimer(actorTimer.getName()), + actorTimer.getCallback(), + actorTimer.getState())) + .then(); + } - /** - * Invokes a timer for a given Actor. - * - * @param actorId Identifier for Actor. - * @param timerName Name of timer being invoked. - * @return Asynchronous void response. - */ - Mono invokeTimer(ActorId actorId, String timerName) { - try { - AbstractActor actor = this.activeActors.getOrDefault(actorId, null); - if (actor == null) { - throw new IllegalArgumentException( - String.format("Could not find actor %s of type %s.", - actorId.toString(), - this.runtimeContext.getActorTypeInformation().getName())); - } + /** + * Internal callback for when Actor is activated. + * + * @param actorId Actor identifier. + * @param actor Actor's instance. + * @return Asynchronous void response. + */ + private Mono onActivatedActor(ActorId actorId, T actor) { + return Mono.fromRunnable(() -> this.activeActors.put(actorId, actor)); + } - ActorTimer actorTimer = actor.getActorTimer(timerName); - if (actorTimer == null) { - throw new IllegalStateException( - String.format("Could not find timer %s for actor %s.", - timerName, - this.runtimeContext.getActorTypeInformation().getName())); - } + /** + * Internal method to actually invoke a reminder. + * + * @param actor Actor that owns the reminder. + * @param reminderName Name of the reminder. + * @param reminderParams Params for the reminder. + * @return Asynchronous void response. + */ + private Mono doReminderInvokation( + Remindable actor, + String reminderName, + ActorReminderParams reminderParams) { + return Mono.fromSupplier(() -> { + if (actor == null) { + throw new IllegalArgumentException("actor is mandatory."); + } + if (reminderName == null) { + throw new IllegalArgumentException("reminderName is mandatory."); + } + if (reminderParams == null) { + throw new IllegalArgumentException("reminderParams is mandatory."); + } - return invokeMethod( - actorId, - ActorMethodContext.CreateForTimer(timerName), - actorTimer.getCallback(), - actorTimer.getState()) - .then(); - } catch (Exception e) { - return Mono.error(e); - } - } + return true; + }).flatMap(x -> { + try { + Object data = this.runtimeContext.getActorSerializer().deserialize( + reminderParams.getData(), + actor.getStateType()); + return actor.receiveReminder( + reminderName, + data, + reminderParams.getDueTime(), + reminderParams.getPeriod()); + } catch (Exception e) { + return Mono.error(e); + } + }).thenReturn(true); + } - /** - * Internal callback for when Actor is activated. - * - * @param actorId Actor identifier. - * @param actor Actor's instance. - * @return Asynchronous void response. - */ - private Mono onActivatedActor(ActorId actorId, T actor) { - this.activeActors.put(actorId, actor); - return Mono.empty(); + /** + * Internal method to actually invoke Actor's method. + * + * @param actorId Identifier for the Actor. + * @param context Method context to be invoked. + * @param methodName Method name to be invoked. + * @param request Input object to be passed in to the invoked method. + * @return Asynchronous void response. + */ + private Mono invokeMethod(ActorId actorId, ActorMethodContext context, String methodName, Object request) { + ActorMethodContext actorMethodContext = context; + if (actorMethodContext == null) { + actorMethodContext = ActorMethodContext.CreateForActor(methodName); } - /** - * Internal method to actually invoke a reminder. - * - * @param actor Actor that owns the reminder. - * @param reminderName Name of the reminder. - * @param reminderParams Params for the reminder. - * @return Asynchronous void response. - */ - private Mono doReminderInvokation( - Remindable actor, - String reminderName, - ActorReminderParams reminderParams) { - try { - Object data = this.runtimeContext.getActorSerializer().deserialize( - reminderParams.getData(), - actor.getStateType()); - return actor.receiveReminder( - reminderName, - data, - reminderParams.getDueTime(), - reminderParams.getPeriod()); - } catch (IOException e) { - return Mono.error(e); - } - } + return this.invoke(actorId, actorMethodContext, actor -> { + try { + // Finds the actor method with the given name and 1 or no parameter. + Method method = this.actorMethods.get(methodName); - /** - * Internal method to actually invoke Actor's method. - * - * @param actorId Identifier for the Actor. - * @param context Method context to be invoked. - * @param methodName Method name to be invoked. - * @param request Input object to be passed in to the invoked method. - * @return Asynchronous void response. - */ - private Mono invokeMethod(ActorId actorId, ActorMethodContext context, String methodName, Object request) { - ActorMethodContext actorMethodContext = context; - if (actorMethodContext == null) { - actorMethodContext = ActorMethodContext.CreateForActor(methodName); - } + if (method.getReturnType().equals(Mono.class)) { + Mono mono = (Mono) invokeMethod(actor, method, request); + if (mono == null) { + return Mono.just(new Object()); + } - return this.invoke(actorId, actorMethodContext, actor -> { + return mono.defaultIfEmpty("").map(r -> { try { - // Finds the actor method with the given name and 1 or no parameter. - Method method = this.actorMethods.get(methodName); - - Object response; + return (Object) this.runtimeContext.getActorSerializer().serializeString(r); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } - if (method.getParameterCount() == 0) { - response = method.invoke(actor); - } else { - // Actor methods must have a one or no parameter, which is guaranteed at this point. - Class inputClass = method.getParameterTypes()[0]; + return Mono.fromSupplier(() -> { + try { + Object response = invokeMethod(actor, method, request); - if ((request != null) && !inputClass.isInstance(request)) { - // If request object is String, we deserialize it. - response = method.invoke( - actor, - this.runtimeContext.getActorSerializer().deserialize(request, inputClass)); - } else { - // If input already of the right type, so we just cast it. - response = method.invoke(actor, inputClass.cast(request)); - } - } + if (response == null) { + return new Object(); + } - if (response == null) { - return Mono.empty(); - } + // Method was not Mono, so we serialize response. + return this.runtimeContext.getActorSerializer().serializeString(response); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } catch (Exception e) { + return Mono.error(e); + } + }).map(r -> r.toString()); + } - if (response instanceof Mono) { - return ((Mono) response).map(r -> { - try { - return this.runtimeContext.getActorSerializer().serializeString(r); - } catch (IOException e) { - throw new RuntimeException(e); - } - }); - } + private Object invokeMethod(AbstractActor actor, Method method, Object request) + throws IllegalAccessException, InvocationTargetException, IOException { + Object response; + if (method.getParameterCount() == 0) { + response = method.invoke(actor); + } else { + // Actor methods must have a one or no parameter, which is guaranteed at this point. + Class inputClass = method.getParameterTypes()[0]; - // Method was not Mono, so we serialize response. - return Mono.just(this.runtimeContext.getActorSerializer().serializeString(response)); - } catch (Exception e) { - return Mono.error(e); - } - }).map(r -> r.toString()); + if ((request != null) && !inputClass.isInstance(request)) { + // If request object is String, we deserialize it. + response = method.invoke( + actor, + this.runtimeContext.getActorSerializer().deserialize(request, inputClass)); + } else { + // If input already of the right type, so we just cast it. + response = method.invoke(actor, inputClass.cast(request)); + } } + return response; + } - /** - * Internal call to invoke a method, timer or reminder for an Actor. - * - * @param actorId Actor identifier. - * @param context Context for the method/timer/reminder call. - * @param func Function to perform the method call. - * @param Expected return type for the function call. - * @return Asynchronous response for the returned object. - */ - private Mono invoke(ActorId actorId, ActorMethodContext context, Function> func) { - try { - AbstractActor actor = this.activeActors.getOrDefault(actorId, null); - if (actor == null) { - throw new IllegalArgumentException( - String.format("Could not find actor %s of type %s.", - actorId.toString(), - this.runtimeContext.getActorTypeInformation().getName())); - } + /** + * Internal call to invoke a method, timer or reminder for an Actor. + * + * @param actorId Actor identifier. + * @param context Context for the method/timer/reminder call. + * @param func Function to perform the method call. + * @param Expected return type for the function call. + * @return Asynchronous response for the returned object. + */ + private Mono invoke(ActorId actorId, ActorMethodContext context, Function> func) { + try { + AbstractActor actor = this.activeActors.getOrDefault(actorId, null); + if (actor == null) { + throw new IllegalArgumentException( + String.format("Could not find actor %s of type %s.", + actorId.toString(), + this.runtimeContext.getActorTypeInformation().getName())); + } - return actor.onPreActorMethodInternal(context).then( - func.apply(actor).flatMap(result -> actor.onPostActorMethodInternal(context).thenReturn(result)) - ); - } catch (Exception e) { - return Mono.error(e); - } + return actor.onPreActorMethodInternal(context) + .then(func.apply(actor)) + .flatMap(result -> actor.onPostActorMethodInternal(context).thenReturn(result)) + .onErrorMap(throwable -> { + actor.resetState(); + return throwable; + }); + } catch (Exception e) { + return Mono.error(e); } + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java index 89df3b5dfe..143c325dbf 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateManager.java @@ -17,301 +17,293 @@ */ public class ActorStateManager { - /** - * Provides states using a state store. - */ - private final DaprStateAsyncProvider stateProvider; - - /** - * Name of the Actor's type. - */ - private final String actorTypeName; - - /** - * Actor's identifier. - */ - private final ActorId actorId; - - /** - * Cache of state changes in this Actor's instance. - */ - private final Map stateChangeTracker; - - /** - * Instantiates a new state manager for the given Actor's instance. - * - * @param stateProvider State store provider. - * @param actorTypeName Name of Actor's type. - * @param actorId Actor's identifier. - */ - ActorStateManager(DaprStateAsyncProvider stateProvider, String actorTypeName, ActorId actorId) { - this.stateProvider = stateProvider; - this.actorTypeName = actorTypeName; - this.actorId = actorId; - this.stateChangeTracker = new HashMap<>(); - } + /** + * Provides states using a state store. + */ + private final DaprStateAsyncProvider stateProvider; + + /** + * Name of the Actor's type. + */ + private final String actorTypeName; + + /** + * Actor's identifier. + */ + private final ActorId actorId; + + /** + * Cache of state changes in this Actor's instance. + */ + private final Map stateChangeTracker; + + /** + * Instantiates a new state manager for the given Actor's instance. + * + * @param stateProvider State store provider. + * @param actorTypeName Name of Actor's type. + * @param actorId Actor's identifier. + */ + ActorStateManager(DaprStateAsyncProvider stateProvider, String actorTypeName, ActorId actorId) { + this.stateProvider = stateProvider; + this.actorTypeName = actorTypeName; + this.actorId = actorId; + this.stateChangeTracker = new HashMap<>(); + } + + /** + * Adds a given key/value to the Actor's state store's cache. + * + * @param stateName Name of the state being added. + * @param value Value to be added. + * @param Type of the object being added. + * @return Asynchronous void operation. + */ + public Mono add(String stateName, T value) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + return null; + }).then(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) + .map(exists -> { + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.UPDATE, value)); + return true; + } + + throw new IllegalStateException("Duplicate cached state: " + stateName); + } - /** - * Adds a given key/value to the Actor's state store's cache. - * - * @param stateName Name of the state being added. - * @param value Value to be added. - * @param Type of the object being added. - * @return Asynchronous void operation. - */ - public Mono add(String stateName, T value) { - try { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.UPDATE, value)); - return Mono.empty(); - } - - throw new IllegalStateException("Duplicate cached state: " + stateName); - } - - return this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) - .flatMap(exists -> { - if (exists) { - throw new IllegalStateException("Duplicate state: " + stateName); - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.ADD, value)); - return Mono.empty(); - }); - } catch (Exception e) { - return Mono.error(e); + if (exists) { + throw new IllegalStateException("Duplicate state: " + stateName); } - } - /** - * Fetches the most recent value for the given state, including cached value. - * - * @param stateName Name of the state. - * @param clazz Class type for the value being fetched. - * @param Type being fetched. - * @return Asynchronous response with fetched object. - */ - public Mono get(String stateName, Class clazz) { - try { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - throw new NoSuchElementException("State is marked for removal: " + stateName); - } - - return Mono.just((T) metadata.value); - } - - return this.stateProvider.load(this.actorTypeName, this.actorId, stateName, clazz) - .switchIfEmpty(Mono.error(new NoSuchElementException("State not found: " + stateName))) - .map(v -> { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, v)); - return (T) v; - }); - } catch (Exception e) { - return Mono.error(e); + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.ADD, value)); + return true; + })) + .then(); + } + + /** + * Fetches the most recent value for the given state, including cached value. + * + * @param stateName Name of the state. + * @param clazz Class type for the value being fetched. + * @param Type being fetched. + * @return Asynchronous response with fetched object. + */ + public Mono get(String stateName, Class clazz) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + throw new NoSuchElementException("State is marked for removal: " + stateName); } - } - /** - * Updates a given key/value pair in the state store's cache. - * - * @param stateName Name of the state being updated. - * @param value Value to be set for given state. - * @param Type of the value being set. - * @return Asynchronous void result. - */ - public Mono set(String stateName, T value) { - try { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - ActorStateChangeKind kind = metadata.kind; - if ((kind == ActorStateChangeKind.NONE) || (kind == ActorStateChangeKind.REMOVE)) { - kind = ActorStateChangeKind.UPDATE; - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(kind, value)); - return Mono.empty(); - } - - return this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) - .map(exists -> { - this.stateChangeTracker.put(stateName, - new StateChangeMetadata(exists ? ActorStateChangeKind.UPDATE : ActorStateChangeKind.ADD, value)); - return exists; - }) - .then(); - } catch (Exception e) { - return Mono.error(e); + return (T) metadata.value; + } + + return (T) null; + }).switchIfEmpty( + this.stateProvider.load(this.actorTypeName, this.actorId, stateName, clazz) + .switchIfEmpty(Mono.error(new NoSuchElementException("State not found: " + stateName))) + .map(v -> { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, v)); + return (T) v; + })); + } + + /** + * Updates a given key/value pair in the state store's cache. + * + * @param stateName Name of the state being updated. + * @param value Value to be set for given state. + * @param Type of the value being set. + * @return Asynchronous void result. + */ + public Mono set(String stateName, T value) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + ActorStateChangeKind kind = metadata.kind; + if ((kind == ActorStateChangeKind.NONE) || (kind == ActorStateChangeKind.REMOVE)) { + kind = ActorStateChangeKind.UPDATE; } - } - /** - * Removes a given state from state store's cache. - * - * @param stateName State being stored. - * @return Asynchronous void result. - */ - public Mono remove(String stateName) { - try { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } - - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - - if (metadata.kind == ActorStateChangeKind.REMOVE) { - return Mono.empty(); - } - - if (metadata.kind == ActorStateChangeKind.ADD) { - this.stateChangeTracker.remove(stateName); - return Mono.empty(); - } - - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); - return Mono.empty(); - } - - return this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) - .filter(exists -> exists) - .map(exists -> { - this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); - return exists; - }) - .then(); - } catch (Exception e) { - return Mono.error(e); + this.stateChangeTracker.put(stateName, new StateChangeMetadata(kind, value)); + return true; + } + + return false; + }).filter(x -> x) + .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName) + .map(exists -> { + this.stateChangeTracker.put(stateName, + new StateChangeMetadata(exists ? ActorStateChangeKind.UPDATE : ActorStateChangeKind.ADD, value)); + return exists; + })) + .then(); + } + + /** + * Removes a given state from state store's cache. + * + * @param stateName State being stored. + * @return Asynchronous void result. + */ + public Mono remove(String stateName) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } + + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + + if (metadata.kind == ActorStateChangeKind.REMOVE) { + return true; } - } - /** - * Checks if a given state exists in state store or cache. - * - * @param stateName State being checked. - * @return Asynchronous boolean result indicating whether state is present. - */ - public Mono contains(String stateName) { - try { - if (stateName == null) { - throw new IllegalArgumentException("State's name cannot be null."); - } + if (metadata.kind == ActorStateChangeKind.ADD) { + this.stateChangeTracker.remove(stateName); + return true; + } - if (this.stateChangeTracker.containsKey(stateName)) { - StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); + return true; + } + + return false; + }) + .filter(x -> x) + .switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)) + .filter(exists -> exists) + .map(exists -> { + this.stateChangeTracker.put(stateName, new StateChangeMetadata(ActorStateChangeKind.REMOVE, null)); + return exists; + }) + .then(); + } + + /** + * Checks if a given state exists in state store or cache. + * + * @param stateName State being checked. + * @return Asynchronous boolean result indicating whether state is present. + */ + public Mono contains(String stateName) { + return Mono.fromSupplier(() -> { + if (stateName == null) { + throw new IllegalArgumentException("State's name cannot be null."); + } - if (metadata.kind == ActorStateChangeKind.REMOVE) { - return Mono.just(false); - } + if (this.stateChangeTracker.containsKey(stateName)) { + StateChangeMetadata metadata = this.stateChangeTracker.get(stateName); - return Mono.just(true); - } + if (metadata.kind == ActorStateChangeKind.REMOVE) { + return Boolean.FALSE; + } - return this.stateProvider.contains(this.actorTypeName, this.actorId, stateName); - } catch (Exception e) { - return Mono.error(e); + return Boolean.TRUE; } - } - /** - * Saves all changes to state store. - * - * @return Asynchronous void result. - */ - public Mono save() { - if (this.stateChangeTracker.isEmpty()) { - return Mono.empty(); + return null; + } + + ).switchIfEmpty(this.stateProvider.contains(this.actorTypeName, this.actorId, stateName)); + } + + /** + * Saves all changes to state store. + * + * @return Asynchronous void result. + */ + public Mono save() { + return Mono.fromSupplier(() -> { + if (this.stateChangeTracker.isEmpty()) { + return null; + } + + List changes = new ArrayList<>(); + List removed = new ArrayList<>(); + for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { + if (tuple.getValue().kind == ActorStateChangeKind.NONE) { + continue; } - List changes = new ArrayList<>(); - List removed = new ArrayList<>(); - for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { - if (tuple.getValue().kind == ActorStateChangeKind.NONE) { - continue; - } - - if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { - removed.add(tuple.getKey()); - } - - changes.add(new ActorStateChange(tuple.getKey(), tuple.getValue().value, tuple.getValue().kind)); + if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { + removed.add(tuple.getKey()); } - return this.stateProvider.apply(this.actorTypeName, this.actorId, changes.toArray(new ActorStateChange[0])) - .then(this.flush()); + changes.add(new ActorStateChange(tuple.getKey(), tuple.getValue().value, tuple.getValue().kind)); + } + + return changes.toArray(new ActorStateChange[0]); + }).flatMap(changes -> this.stateProvider.apply(this.actorTypeName, this.actorId, changes)) + .then(Mono.fromRunnable(() -> this.flush())); + } + + /** + * Clears all changes not yet saved to state store. + */ + public void clear() { + this.stateChangeTracker.clear(); + } + + /** + * Commits the current cached values after successful save. + */ + private void flush() { + for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { + String stateName = tuple.getKey(); + if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { + this.stateChangeTracker.remove(stateName); + } else { + StateChangeMetadata metadata = new StateChangeMetadata(ActorStateChangeKind.NONE, tuple.getValue().value); + this.stateChangeTracker.put(stateName, metadata); + } } + } + + /** + * Internal class to represent value and change kind. + */ + private static final class StateChangeMetadata { /** - * Clears all changes not yet saved to state store. - * - * @return + * Kind of change cached. */ - public Mono clear() { - this.stateChangeTracker.clear(); - return Mono.empty(); - } + private final ActorStateChangeKind kind; /** - * Commits the current cached values after successful save. - * - * @return + * Value cached. */ - private Mono flush() { - for (Map.Entry tuple : this.stateChangeTracker.entrySet()) { - String stateName = tuple.getKey(); - if (tuple.getValue().kind == ActorStateChangeKind.REMOVE) { - this.stateChangeTracker.remove(stateName); - } else { - StateChangeMetadata metadata = new StateChangeMetadata(ActorStateChangeKind.NONE, tuple.getValue().value); - this.stateChangeTracker.put(stateName, metadata); - } - } - - return Mono.empty(); - } + private final Object value; /** - * Internal class to represent value and change kind. + * Creates a new instance of the metadata on state change. + * + * @param kind Kind of change. + * @param value Value to be set. */ - private static final class StateChangeMetadata { - - /** - * Kind of change cached. - */ - private final ActorStateChangeKind kind; - - /** - * Value cached. - */ - private final Object value; - - /** - * Creates a new instance of the metadata on state change. - * - * @param kind Kind of change. - * @param value Value to be set. - */ - private StateChangeMetadata(ActorStateChangeKind kind, Object value) { - this.kind = kind; - this.value = value; - } + private StateChangeMetadata(ActorStateChangeKind kind, Object value) { + this.kind = kind; + this.value = value; } + } } diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java index 38cb87e2b4..e803b87b09 100644 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java +++ b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorStateSerializer.java @@ -58,20 +58,20 @@ public T deserialize(Object value, Class clazz) throws IOException { } /** - * Extracts the response object from the Actor's method result. + * Extracts the response object from a JSON Payload where data is in "data" attribute. * - * @param response String returned by API. + * @param payload JSON payload containing "data". * @param clazz Expected response class. * @param Expected response type. * @return Response object, null or RuntimeException. */ - public T unwrapMethodResponse(final String response, Class clazz) { - if (response == null) { + public T unwrapData(final String payload, Class clazz) { + if (payload == null) { return null; } try { - JsonNode root = OBJECT_MAPPER.readTree(response); + JsonNode root = OBJECT_MAPPER.readTree(payload); if (root == null) { return null; } @@ -96,17 +96,16 @@ public T unwrapMethodResponse(final String response, Class clazz) { /** * Builds the request to invoke an API for Actors. * - * @param request Request object for the original Actor's method. - * @param Type for the original Actor's method request. + * @param object Object to be serialized and wrapped into the "data" attribute in a JSON object. * @return String to be sent to Dapr's API. * @throws IOException In case it cannot generate String. */ - public String wrapMethodRequest(final T request) throws IOException { - if (request == null) { + public String wrapData(final Object object) throws IOException { + if (object == null) { return null; } - byte[] data = this.serialize(request); + byte[] data = this.serialize(object); try (Writer writer = new StringWriter()) { JsonGenerator generator = JSON_FACTORY.createGenerator(writer); diff --git a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTimerParams.java b/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTimerParams.java deleted file mode 100644 index d2fdaad40d..0000000000 --- a/sdk-actors/src/main/java/io/dapr/actors/runtime/ActorTimerParams.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright (c) Microsoft Corporation. - * Licensed under the MIT License. - */ - -package io.dapr.actors.runtime; - -import java.time.Duration; - -/** - * Represents the timer set on an Actor, to be called once after due time and then every period. - */ -final class ActorTimerParams { - - /** - * Name of the method to be called for this timer. - */ - private String callback; - - /** - * Data to be sent in the timer. - */ - private String data; - - /** - * Due time for the timer's first trigger. - */ - private Duration dueTime; - - /** - * Period at which the timer will be triggered. - */ - private Duration period; - - /** - * Instantiates new params for Actor Timer. - * - * @param callback The name of the method to be called for this timer. - * @param data Information to be used by the callback method - * @param dueTime The time when timer is first due. - * @param period The periodic time when timer will be invoked. - */ - ActorTimerParams(String callback, - String data, - Duration dueTime, - Duration period) { - this.callback = callback; - this.data = data; - this.dueTime = dueTime; - this.period = period; - } - - /** - * Gets the name of the method for this Timer. - * - * @return The name of the method for this timer. - */ - public String getCallback() { - return this.callback; - } - - /** - * Gets the time when timer is first due. - * - * @return Time as Duration when timer is first due. - */ - public Duration getDueTime() { - return this.dueTime; - } - - /** - * Gets the periodic time when timer will be invoked. - * - * @return Periodic time as Duration when timer will be invoked. - */ - public Duration getPeriod() { - return this.period; - } - - /** - * Gets data to be used by the callback method, or null. - * - * @return Data to be used by the callback method, or null. - */ - public String getData() { - return this.data; - } - -} \ No newline at end of file diff --git a/sdk-actors/src/test/java/io/dapr/actors/client/ActorProxyForTestsImpl.java b/sdk-actors/src/test/java/io/dapr/actors/client/ActorProxyForTestsImpl.java new file mode 100644 index 0000000000..df0006841b --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/client/ActorProxyForTestsImpl.java @@ -0,0 +1,17 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.actors.client; + +import io.dapr.actors.ActorId; +import io.dapr.actors.runtime.ActorStateSerializer; +import io.dapr.client.DaprClient; + +public class ActorProxyForTestsImpl extends ActorProxyImpl { + + public ActorProxyForTestsImpl(String actorType, ActorId actorId, ActorStateSerializer serializer, DaprClient daprClient) { + super(actorType, actorId, serializer, daprClient); + } +} diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java index 0366552864..f79dfca8a3 100644 --- a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorManagerTest.java @@ -15,7 +15,9 @@ import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** * Unit tests for Actor Manager @@ -66,7 +68,7 @@ public MyActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { 2, Duration.ofSeconds(1), Duration.ofSeconds(1) - ); + ).block(); } @Override @@ -129,15 +131,15 @@ public void invokeReminderBeforeActivate() throws Exception { @Test public void activateThenInvokeReminder() throws Exception { ActorId actorId = newActorId(); - this.manager.activateActor(actorId); + this.manager.activateActor(actorId).block(); this.manager.invokeReminder(actorId, "myremind", createReminderParams("hello")).block(); } @Test(expected = IllegalArgumentException.class) public void activateDeactivateThenInvokeReminder() throws Exception { ActorId actorId = newActorId(); - this.manager.activateActor(actorId); - this.manager.deactivateActor(actorId); + this.manager.activateActor(actorId).block(); + this.manager.deactivateActor(actorId).block();; this.manager.invokeReminder(actorId, "myremind", createReminderParams("hello")).block(); } @@ -189,12 +191,19 @@ private static String executeSayMethod(String something) { } private static ActorRuntimeContext createContext(Class clazz) { + DaprClient daprClient = mock(DaprClient.class); + + when(daprClient.registerActorTimer(any(), any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.registerActorReminder(any(), any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.unregisterActorTimer(any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.unregisterActorReminder(any(), any(), any())).thenReturn(Mono.empty()); + return new ActorRuntimeContext( mock(ActorRuntime.class), new ActorStateSerializer(), new DefaultActorFactory(), ActorTypeInformation.create(clazz), - mock(DaprClient.class), + daprClient, mock(DaprStateAsyncProvider.class) ); } diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorStatefulTest.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorStatefulTest.java new file mode 100644 index 0000000000..73a38137b0 --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/ActorStatefulTest.java @@ -0,0 +1,660 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.actors.runtime; + +import io.dapr.actors.ActorId; +import io.dapr.actors.client.ActorProxy; +import io.dapr.actors.client.ActorProxyForTestsImpl; +import io.dapr.client.DaprClient; +import org.junit.Assert; +import org.junit.Test; +import reactor.core.publisher.Mono; + +import java.io.IOException; +import java.nio.charset.IllegalCharsetNameException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ActorStatefulTest { + + private static final AtomicInteger ACTOR_ID_COUNT = new AtomicInteger(); + + private static final Collection DEACTIVATED_ACTOR_IDS = Collections.synchronizedList(new ArrayList<>()); + + private final ActorRuntimeContext context = createContext(); + + private ActorManager manager = new ActorManager<>(context); + + public interface MyActor { + Mono isActive(); + + MyMethodContext getPreCallMethodContext(); + + MyMethodContext getPostCallMethodContext(); + + Mono unregisterTimerAndReminder(); + + Mono incrementAndGetCount(int increment) throws Exception; + + Mono getCountButThrowsException(); + + Mono addMessage(String message); + + Mono setMessage(String message); + + Mono getMessage(); + + Mono hasMessage(); + + Mono deleteMessage(); + + Mono forceDuplicateException(); + + Mono forcePartialChange(); + + Mono throwsWithoutSaving(); + + Mono setMethodContext(MyMethodContext context); + + Mono getMethodContext(); + + String getIdString(); + } + + @ActorType(Name = "MyActor") + public static class MyActorImpl extends AbstractActor implements MyActor, Actor, Remindable { + + private final ActorId id; + + private boolean activated; + + private MyMethodContext preMethodCalled; + + private MyMethodContext postMethodCalled; + + public MyActorImpl(ActorRuntimeContext runtimeContext, ActorId id) { + super(runtimeContext, id); + this.id = id; + this.activated = true; + } + + @Override + public Mono isActive() { + return Mono.fromSupplier(() -> this.activated); + } + + @Override + public Mono onActivate() { + return Mono + .fromRunnable(() -> this.activated = true) + .then(super.registerActorTimer( + "mytimer", + "hasMessage", + null, + Duration.ofSeconds(1), + Duration.ofSeconds(1))) + .then(super.registerReminder( + "myreminder", + null, + Duration.ofSeconds(1), + Duration.ofSeconds(1) + )); + } + + @Override + public Mono onDeactivate() { + return Mono.fromRunnable(() -> DEACTIVATED_ACTOR_IDS.add(this.id.toString())); + } + + @Override + public Mono onPreActorMethod(ActorMethodContext context) { + // Only keep the first one to make sure we can validate it via another method invocation. + return Mono.fromRunnable(() -> { + this.preMethodCalled = this.preMethodCalled != null ? this.preMethodCalled : new MyMethodContext() + .setName(context.getMethodName()) + .setType(context.getCallType().toString()); + }); + } + + @Override + public Mono onPostActorMethod(ActorMethodContext context) { + // Only keep the first one to make sure we can validate it via another method invocation. + return Mono.fromRunnable(() -> { + this.postMethodCalled = this.postMethodCalled != null ? this.postMethodCalled : new MyMethodContext() + .setName(context.getMethodName()) + .setType(context.getCallType().toString()); + }); + } + + @Override + public MyMethodContext getPreCallMethodContext() { + return this.preMethodCalled; + } + + @Override + public MyMethodContext getPostCallMethodContext() { + return this.postMethodCalled; + } + + @Override + public Mono unregisterTimerAndReminder() { + return super.unregisterReminder("UnknownReminder") + .then(super.unregisterTimer("UnknownTimer")) + .then(super.unregisterReminder("myreminder")) + .then(super.unregisterTimer("mytimer")); + } + + @Override + public Mono incrementAndGetCount(int increment) { + return Mono.fromRunnable(() -> { + if (increment == 0) { + // Artificial exception case for testing. + throw new NumberFormatException("increment cannot be zero."); + } + }) + .then(super.getActorStateManager().contains("counter")) + .flatMap(contains -> { + if (!contains) { + return Mono.just(0); + } + + return super.getActorStateManager().get("counter", int.class); + }) + .map(count -> count + increment) + .flatMap(count -> super.getActorStateManager().set("counter", count).thenReturn(count)); + } + + @Override + public Mono getCountButThrowsException() { + return super.getActorStateManager().get("counter_WRONG_NAME", int.class); + } + + @Override + public Mono addMessage(String message) { + return super.getActorStateManager().add("message", message); + } + + @Override + public Mono setMessage(String message) { + return super.getActorStateManager().set("message", message).thenReturn(executeSayMethod(message)); + } + + @Override + public Mono getMessage() { + return super.getActorStateManager().get("message", String.class); + } + + @Override + public Mono hasMessage() { + return super.getActorStateManager().contains("message"); + } + + @Override + public Mono deleteMessage() { + return super.getActorStateManager().remove("message"); + } + + @Override + public Mono forceDuplicateException() { + // Second add should throw exception. + return super.getActorStateManager().add("message", "anything") + .then(super.getActorStateManager().add("message", "something else")); + } + + @Override + public Mono forcePartialChange() { + return super.getActorStateManager().add("message", "first message") + .then(super.saveState()) + .then(super.getActorStateManager().add("message", "second message")); + } + + @Override + public Mono throwsWithoutSaving() { + return super.getActorStateManager().add("message", "first message") + .then(Mono.error(new IllegalCharsetNameException("random"))); + } + + @Override + public Mono setMethodContext(MyMethodContext context) { + return super.getActorStateManager().set("context", context); + } + + @Override + public Mono getMethodContext() { + return super.getActorStateManager().get("context", MyMethodContext.class); + } + + // Blocking methods are also supported for Actors. Mono is not required. + @Override + public String getIdString() { + return this.id.toString(); + } + + @Override + public Class getStateType() { + // Remindable type. + return String.class; + } + + @Override + public Mono receiveReminder(String reminderName, String state, Duration dueTime, Duration period) { + return Mono.empty(); + } + } + + // Class used to validate serialization/deserialization + public static class MyMethodContext { + + private String type; + + private String name; + + public String getType() { + return type; + } + + public MyMethodContext setType(String type) { + this.type = type; + return this; + } + + public String getName() { + return name; + } + + public MyMethodContext setName(String name) { + this.name = name; + return this; + } + } + + @Test + public void happyGetSetDeleteContains() { + ActorProxy proxy = newActorProxy(); + Assert.assertEquals( + proxy.getActorId().toString(), proxy.invokeActorMethod("getIdString", String.class).block()); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + proxy.invokeActorMethod("setMessage", "hello world").block(); + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + Assert.assertEquals( + "hello world", proxy.invokeActorMethod("getMessage", String.class).block()); + + Assert.assertEquals( + executeSayMethod("hello world"), + proxy.invokeActorMethod("setMessage", "hello world", String.class).block()); + + proxy.invokeActorMethod("deleteMessage").block(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test(expected = IllegalStateException.class) + public void lazyGet() { + ActorProxy proxy = newActorProxy(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + proxy.invokeActorMethod("setMessage", "first message").block(); + + // Creates the mono plan but does not call it yet. + Mono getMessageCall = proxy.invokeActorMethod("getMessage", String.class); + + proxy.invokeActorMethod("deleteMessage").block(); + + // Call should fail because the message was deleted. + getMessageCall.block(); + } + + @Test + public void lazySet() { + ActorProxy proxy = newActorProxy(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Creates the mono plan but does not call it yet. + Mono setMessageCall = proxy.invokeActorMethod("setMessage", "first message"); + + // No call executed yet, so message should not be set. + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + setMessageCall.block(); + + // Now the message has been set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test + public void lazyContains() { + ActorProxy proxy = newActorProxy(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Creates the mono plan but does not call it yet. + Mono hasMessageCall = proxy.invokeActorMethod("hasMessage", Boolean.class); + + // Sets the message. + proxy.invokeActorMethod("setMessage", "hello world").block(); + + // Now we check if message is set. + hasMessageCall.block(); + + // Now the message should be set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test + public void lazyDelete() { + ActorProxy proxy = newActorProxy(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + proxy.invokeActorMethod("setMessage", "first message").block(); + + // Message is set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Created the mono plan but does not execute it yet. + Mono deleteMessageCall = proxy.invokeActorMethod("deleteMessage"); + + // Message is still set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + deleteMessageCall.block(); + + // Now message is not set. + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test + public void lazyAdd() { + ActorProxy proxy = newActorProxy(); + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + proxy.invokeActorMethod("setMessage", "first message").block(); + + // Message is set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Created the mono plan but does not execute it yet. + Mono addMessageCall = proxy.invokeActorMethod("addMessage", "second message"); + + // Message is still set. + Assert.assertEquals("first message", + proxy.invokeActorMethod("getMessage", String.class).block()); + + // Delete message + proxy.invokeActorMethod("deleteMessage").block(); + + // Should work since previous message was deleted. + addMessageCall.block(); + + // New message is still set. + Assert.assertEquals("second message", + proxy.invokeActorMethod("getMessage", String.class).block()); + } + + @Test + public void onActivateAndOnDeactivate() { + ActorProxy proxy = newActorProxy(); + + Assert.assertTrue(proxy.invokeActorMethod("isActive", Boolean.class).block()); + Assert.assertFalse(DEACTIVATED_ACTOR_IDS.contains(proxy.getActorId().toString())); + + proxy.invokeActorMethod("hasMessage", Boolean.class).block(); + + this.manager.deactivateActor(proxy.getActorId()).block(); + + Assert.assertTrue(DEACTIVATED_ACTOR_IDS.contains(proxy.getActorId().toString())); + } + + @Test + public void onPreMethodAndOnPostMethod() { + ActorProxy proxy = newActorProxy(); + + proxy.invokeActorMethod("hasMessage", Boolean.class).block(); + + MyMethodContext preContext = + proxy.invokeActorMethod("getPreCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("hasMessage", preContext.getName()); + Assert.assertEquals(ActorCallType.ACTOR_INTERFACE_METHOD.toString(), preContext.getType()); + + MyMethodContext postContext = + proxy.invokeActorMethod("getPostCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("hasMessage", postContext.getName()); + Assert.assertEquals(ActorCallType.ACTOR_INTERFACE_METHOD.toString(), postContext.getType()); + } + + @Test + public void invokeTimer() { + ActorProxy proxy = newActorProxy(); + + this.manager.invokeTimer(proxy.getActorId(), "mytimer").block(); + + MyMethodContext preContext = + proxy.invokeActorMethod("getPreCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("mytimer", preContext.getName()); + Assert.assertEquals(ActorCallType.TIMER_METHOD.toString(), preContext.getType()); + + MyMethodContext postContext = + proxy.invokeActorMethod("getPostCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("mytimer", postContext.getName()); + Assert.assertEquals(ActorCallType.TIMER_METHOD.toString(), postContext.getType()); + } + + @Test(expected = IllegalArgumentException.class) + public void invokeTimerAfterDeactivate() { + ActorProxy proxy = newActorProxy(); + + this.manager.deactivateActor(proxy.getActorId()).block(); + + this.manager.invokeTimer(proxy.getActorId(), "mytimer").block(); + } + + @Test(expected = IllegalStateException.class) + public void invokeTimerAfterUnregister() { + ActorProxy proxy = newActorProxy(); + + proxy.invokeActorMethod("unregisterTimerAndReminder").block(); + + this.manager.invokeTimer(proxy.getActorId(), "mytimer").block(); + } + + @Test(expected = IllegalStateException.class) + public void invokeUnknownTimer() { + ActorProxy proxy = newActorProxy(); + + this.manager.invokeTimer(proxy.getActorId(), "unknown").block(); + } + + @Test + public void invokeReminder() throws Exception { + ActorProxy proxy = newActorProxy(); + + String params = createReminderParams("anything"); + + this.manager.invokeReminder(proxy.getActorId(), "myreminder", params).block(); + + MyMethodContext preContext = + proxy.invokeActorMethod("getPreCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("myreminder", preContext.getName()); + Assert.assertEquals(ActorCallType.REMINDER_METHOD.toString(), preContext.getType()); + + MyMethodContext postContext = + proxy.invokeActorMethod("getPostCallMethodContext", MyMethodContext.class).block(); + Assert.assertEquals("myreminder", postContext.getName()); + Assert.assertEquals(ActorCallType.REMINDER_METHOD.toString(), postContext.getType()); + } + + @Test(expected = IllegalArgumentException.class) + public void invokeReminderAfterDeactivate() throws Exception { + ActorProxy proxy = newActorProxy(); + + this.manager.deactivateActor(proxy.getActorId()).block(); + + String params = createReminderParams("anything"); + + this.manager.invokeReminder(proxy.getActorId(), "myreminder", params).block(); + } + + @Test + public void classTypeRequestResponseInStateStore() { + ActorProxy proxy = newActorProxy(); + + MyMethodContext expectedContext = new MyMethodContext().setName("MyName").setType("MyType"); + + proxy.invokeActorMethod("setMethodContext", expectedContext).block(); + MyMethodContext context = proxy.invokeActorMethod("getMethodContext", MyMethodContext.class).block(); + + Assert.assertEquals(expectedContext.getName(), context.getName()); + Assert.assertEquals(expectedContext.getType(), context.getType()); + } + + @Test + public void intTypeRequestResponseInStateStore() { + ActorProxy proxy = newActorProxy(); + + Assert.assertEquals(1, (int)proxy.invokeActorMethod("incrementAndGetCount", 1, int.class).block()); + Assert.assertEquals(6, (int)proxy.invokeActorMethod("incrementAndGetCount", 5, int.class).block()); + } + + @Test(expected = NumberFormatException.class) + public void intTypeWithMethodException() { + ActorProxy proxy = newActorProxy(); + + // Zero is a magic input that will make method throw an exception. + proxy.invokeActorMethod("incrementAndGetCount", 0, int.class).block(); + } + + @Test(expected = IllegalStateException.class) + public void intTypeWithRuntimeException() { + ActorProxy proxy = newActorProxy(); + + proxy.invokeActorMethod("getCountButThrowsException", int.class).block(); + } + + @Test(expected = IllegalStateException.class) + public void actorRuntimeException() { + ActorProxy proxy = newActorProxy(); + + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + proxy.invokeActorMethod("forceDuplicateException").block(); + } + + @Test(expected = IllegalCharsetNameException.class) + public void actorMethodException() { + ActorProxy proxy = newActorProxy(); + + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + proxy.invokeActorMethod("throwsWithoutSaving").block(); + + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test + public void rollbackChanges() { + ActorProxy proxy = newActorProxy(); + + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Runs a method that will add one message but fail because tries to add a second one. + proxy.invokeActorMethod("forceDuplicateException") + .onErrorResume(throwable -> Mono.empty()) + .block(); + + // No message is set + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + } + + @Test + public void partialChanges() { + ActorProxy proxy = newActorProxy(); + + Assert.assertFalse(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // Runs a method that will add one message, commit but fail because tries to add a second one. + proxy.invokeActorMethod("forcePartialChange") + .onErrorResume(throwable -> Mono.empty()) + .block(); + + // Message is set. + Assert.assertTrue(proxy.invokeActorMethod("hasMessage", Boolean.class).block()); + + // It is first message and not the second due to a save() in the middle but an exception in the end. + Assert.assertEquals("first message", + proxy.invokeActorMethod("getMessage", String.class).block()); + } + + private ActorProxy newActorProxy() { + ActorId actorId = newActorId(); + + // Mock daprClient for ActorProxy only, not for runtime. + DaprClient daprClient = mock(DaprClient.class); + + when(daprClient.invokeActorMethod( + eq(context.getActorTypeInformation().getName()), + eq(actorId.toString()), + any(), + any())) + .thenAnswer(invocationOnMock -> + this.manager.invokeMethod( + new ActorId(invocationOnMock.getArgument(1, String.class)), + invocationOnMock.getArgument(2, String.class), + context.getActorSerializer().unwrapData( + invocationOnMock.getArgument(3, String.class), String.class)) + .map(s -> { + try { + return context.getActorSerializer().wrapData(s); + } catch (Exception e) { + throw new RuntimeException(e); + } + })); + + this.manager.activateActor(actorId).block(); + + return new ActorProxyForTestsImpl( + context.getActorTypeInformation().getName(), + actorId, + new ActorStateSerializer(), + daprClient); + } + + private String createReminderParams(String data) throws IOException { + ActorReminderParams params = new ActorReminderParams(data, Duration.ofSeconds(1), Duration.ofSeconds(1)); + return this.context.getActorSerializer().serializeString(params); + } + + private static ActorId newActorId() { + return new ActorId(Integer.toString(ACTOR_ID_COUNT.incrementAndGet())); + } + + private static String executeSayMethod(String something) { + return "Said: " + (something == null ? "" : something); + } + + private static ActorRuntimeContext createContext() { + DaprClient daprClient = mock(DaprClient.class); + + when(daprClient.registerActorTimer(any(), any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.registerActorReminder(any(), any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.unregisterActorTimer(any(), any(), any())).thenReturn(Mono.empty()); + when(daprClient.unregisterActorReminder(any(), any(), any())).thenReturn(Mono.empty()); + + return new ActorRuntimeContext( + mock(ActorRuntime.class), + new ActorStateSerializer(), + new DefaultActorFactory(), + ActorTypeInformation.create(MyActorImpl.class), + daprClient, + new DaprInMemoryStateProvider(new ActorStateSerializer()) + ); + } +} diff --git a/sdk-actors/src/test/java/io/dapr/actors/runtime/DaprInMemoryStateProvider.java b/sdk-actors/src/test/java/io/dapr/actors/runtime/DaprInMemoryStateProvider.java new file mode 100644 index 0000000000..f0d74c6c31 --- /dev/null +++ b/sdk-actors/src/test/java/io/dapr/actors/runtime/DaprInMemoryStateProvider.java @@ -0,0 +1,78 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.actors.runtime; + +import io.dapr.actors.ActorId; +import io.dapr.client.DaprClient; +import reactor.core.publisher.Mono; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +/** + * Fake state provider for tests in Actors - data is kept in memory only. + */ +public class DaprInMemoryStateProvider extends DaprStateAsyncProvider { + + private static final Map stateStore = new HashMap<>(); + + private final ActorStateSerializer serializer; + + DaprInMemoryStateProvider(ActorStateSerializer serializer) { + super(null, null); + this.serializer = serializer; + } + + @Override + Mono load(String actorType, ActorId actorId, String stateName, Class clazz) { + return Mono.fromSupplier(() -> { + try { + String stateId = this.buildId(actorType, actorId, stateName); + if (!stateStore.containsKey(stateId)) { + throw new IllegalStateException("State not found."); + } + + return this.serializer.deserialize(this.stateStore.get(stateId), clazz); + } catch (IOException e) { + throw new RuntimeException(e); + } + }); + } + + @Override + Mono contains(String actorType, ActorId actorId, String stateName) { + return Mono.fromSupplier(() -> stateStore.containsKey(this.buildId(actorType, actorId, stateName))); + } + + @Override + Mono apply(String actorType, ActorId actorId, ActorStateChange... stateChanges) { + return Mono.fromRunnable(() -> { + try { + for (ActorStateChange stateChange : stateChanges) { + String stateId = buildId(actorType, actorId, stateChange.getStateName()); + switch (stateChange.getChangeKind()) { + case REMOVE: + stateStore.remove(stateId); + break; + case ADD: + case UPDATE: + byte[] raw = this.serializer.serialize(stateChange.getValue()); + stateStore.put(stateId, raw); + break; + } + } + + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + } + + private static final String buildId(String actorType, ActorId actorId, String stateName) { + return String.format("%s||%s||%s", actorType, actorId.toString(), stateName); + } +}