Feat. Added Effects, Pipes, and Forwards

Author: sleipnir

README.md

@@ -7,8 +7,8 @@ Python User Language Support for [Spawn](https://github.com/eigr/spawn).
 2. [Getting Started](#getting-started)
 3. [Advanced Use Cases](#advanced-use-cases)
    - [Types of Actors](#types-of-actors)
-   - [Side Effects](#side-effects)
    - [Broadcast](#broadcast)
+   - [Side Effects](#side-effects)
    - [Forward](#forward)
    - [Pipe](#pipe)
 4. [Using Actors](#using-actors)
@@ -163,9 +163,6 @@ First we need to understand how the various types of actors available in Spawn b
 
 * **Pooled Actors**: Pooled Actors, as the name suggests, are a collection of actors that are grouped under the same name assigned to them at compile time. Pooled actors are generally used when higher performance is needed and are also recommended for handling serverless loads.
 
-### Side Effects
-TODO
-
 ### Broadcast
 
 Actors in Spawn can subscribe to a thread and receive, as well as broadcast, events for a given thread.
@@ -221,11 +218,107 @@ def set_language(request: Request, ctx: Context) -> Value:
     return Value().of(reply, ctx.state).reply()
 ```
 
+### Side Effects
+
+Actors can also emit side effects to other Actors as part of their response.
+See an example:
+
+```python
+from domain.domain_pb2 import State, Request, Reply
+from spawn.eigr.functions.actors.api.actor import Actor
+from spawn.eigr.functions.actors.api.settings import ActorSettings
+from spawn.eigr.functions.actors.api.context import Context
+from spawn.eigr.functions.actors.api.value import Value
+from spawn.eigr.functions.actors.api.workflows.effect import Effect
+
+
+actor = Actor(settings=ActorSettings(name="joe", stateful=True))
+
+
+@actor.timer_action(every=1000)
+def hi(ctx: Context) -> Value:
+    new_state = None
+    request = Request()
+    request.language = "python"
+
+    effect: Effect = Effect(action="setLanguage", payload=request,
+                            system="spawn-system", actor="mike", parent="abs_actor")
+
+    if not ctx.state:
+        new_state = State()
+        new_state.languages.append("python")
+    else:
+        new_state = ctx.state
+
+    return Value()\
+        .effect(effect)\
+        .state(new_state)\
+        .noreply()
+
+```
+
+Side effects such as broadcast are not part of the response flow to the caller. They are request-asynchronous events that are emitted after the Actor's state has been saved in memory.
+
 ### Forward
-TODO
+
+Actors can route some actions to other actors as part of their response. For example, sometimes you may want another Actor to be responsible for processing a message that another Actor has received. We call this forwarding and it occurs when we want to forward the input argument of a request that a specific Actor has received to the input of an action in another Actor.
+
+See an example:
+
+```python
+from domain.domain_pb2 import Request
+
+from spawn.eigr.functions.actors.api.actor import Actor
+from spawn.eigr.functions.actors.api.settings import ActorSettings
+from spawn.eigr.functions.actors.api.context import Context
+from spawn.eigr.functions.actors.api.value import Value
+from spawn.eigr.functions.actors.api.workflows.forward import Forward
+
+actor = Actor(settings=ActorSettings(name="joe", stateful=True))
+
+@actor.action("setLanguage")
+def set_language(request: Request, ctx: Context) -> Value:
+    return Value()\
+        .forward(Forward("mike", "setLanguage"))\
+        .reply()
+```
 
 ### Pipe
-TODO
+
+Similarly, sometimes we want to chain a request through several different processes. For example forwarding an actor's computational output as another actor's input. There is this type of routing we call Pipe, as the name suggests, a pipe forwards what would be the response of the received request to the input of another Action in another Actor.
+In the end, just like in a Forward, it is the response of the last Actor in the chain of routing to the original caller.
+
+Example:
+
+```python
+from domain.domain_pb2 import State, Request, Reply
+
+from spawn.eigr.functions.actors.api.actor import Actor
+from spawn.eigr.functions.actors.api.settings import ActorSettings
+from spawn.eigr.functions.actors.api.context import Context
+from spawn.eigr.functions.actors.api.value import Value
+from spawn.eigr.functions.actors.api.workflows.pipe import Pipe
+
+actor = Actor(settings=ActorSettings(name="joe", stateful=True))
+
+@actor.action("setLanguage")
+def set_language(request: Request, ctx: Context) -> Value:
+    reply = Reply()
+    reply.language = "python"
+
+    if not ctx.state:
+        new_state = State()
+        new_state.languages.append("python")
+    else:
+        new_state = ctx.state
+
+    return Value()\
+        .response(reply)\
+        .pipe(Pipe("mike", "setLanguage"))\
+        .reply()
+```
+
+Forwards and pipes do not have an upper thread limit other than the request timeout.
 
 ## Using Actors
 

example/joe.py

@@ -9,6 +9,7 @@
 from spawn.eigr.functions.actors.api.value import Value
 from spawn.eigr.functions.actors.api.workflows.broadcast import Broadcast
 
+
 actor = Actor(settings=ActorSettings(
     name="joe", stateful=True, channel="test"))
 
@@ -18,6 +19,7 @@ def hi(ctx: Context) -> Value:
     new_state = None
     request = Request()
     request.language = "python"
+
     broadcast = Broadcast()
     broadcast.channel = "test"
     broadcast.action_name = "setLanguage"

spawn/eigr/functions/actors/api/value.py

@@ -1,7 +1,8 @@
 # pylint: disable=too-few-public-methods
 # pylint: disable=arguments-differ
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from enum import Enum
+from typing import List
 
 from spawn.eigr.functions.actors.api.metadata import Metadata
 from spawn.eigr.functions.actors.api.workflows.broadcast import Broadcast
@@ -22,7 +23,7 @@ class Value():
     __response = None
     __metadata: Metadata = None
     __broadcast: Broadcast = None
-    __effect: Effect = None
+    __effects: List[Effect] = field(default_factory=list)
     __forward: Forward = None
     __pipe: Pipe = None
     __reply_kind: ReplyKind = ReplyKind.REPLY
@@ -42,8 +43,8 @@ def get_broadcast(self):
     def get_broadcast(self):
         return self.__broadcast
 
-    def get_effect(self):
-        return self.__effect
+    def get_effects(self):
+        return self.__effects
 
     def get_forward(self):
         return self.__forward
@@ -76,14 +77,22 @@ def broadcast(self, broadcast: Broadcast):
         return self
 
     def effect(self, effect: Effect):
-        self.__effect = effect
+        self.__effects.append(effect)
         return self
 
     def forward(self, forward: Forward):
+        if self.__pipe:
+            raise Exception(
+                "You can only use either Pipe or Forward never both")
+
         self.__forward = forward
         return self
 
     def pipe(self, pipe: Pipe):
+        if self.__forward:
+            raise Exception(
+                "You can only use either Forward or Pipe never both")
+
         self.__pipe = pipe
         return self
 

spawn/eigr/functions/actors/api/workflows/effect.py

@@ -1,4 +1,8 @@
 
 class Effect:
-    def __init__(self):
-        self
+    def __init__(self, action: str, system: str, actor: str, parent: str, payload: any = None):
+        self.action = action
+        self.actor = actor
+        self.system = system
+        self.parent = parent
+        self.payload = payload

spawn/eigr/functions/actors/api/workflows/forward.py

@@ -1,4 +1,5 @@
 
 class Forward:
-    def __init__(self):
-        pass
+    def __init__(self, actor: str, action: str):
+        self.actor = actor
+        self.action = action

spawn/eigr/functions/actors/api/workflows/pipe.py

@@ -1,4 +1,5 @@
 
 class Pipe:
-    def __init__(self):
-        pass
+    def __init__(self, actor: str, action: str):
+        self.actor = actor
+        self.action = action

spawn/eigr/functions/actors/internal/controller.py

@@ -8,10 +8,12 @@
 
 from spawn.eigr.functions.actors.api.actor import Actor as ActorEntity
 from spawn.eigr.functions.actors.api.actor import ActorHandler
+
 from spawn.eigr.functions.actors.api.context import Context as ActorContext
 from spawn.eigr.functions.actors.api.value import Value, ReplyKind
 from spawn.eigr.functions.actors.api.settings import Kind as ActorKind
 from spawn.eigr.functions.actors.internal.client import SpawnClient
+from spawn.eigr.functions.actors.api.workflows.effect import Effect
 
 from spawn.eigr.functions.protocol.actors.actor_pb2 import (
     Actor,
@@ -32,13 +34,18 @@
 from spawn.eigr.functions.protocol.actors.protocol_pb2 import (
     ActorInvocation,
     ActorInvocationResponse,
+    Broadcast,
+    Forward,
     Context,
+    InvocationRequest,
     Noop,
+    Pipe,
     RegistrationRequest,
-    ServiceInfo,
-    Broadcast
+    ServiceInfo
 )
 
+from spawn.eigr.functions.protocol.actors.protocol_pb2 import SideEffect as SpawnEffect
+
 from google.protobuf import symbol_database as _symbol_database
 from google.protobuf.any_pb2 import Any as AnyProto
 
@@ -100,9 +107,61 @@ def handle_response(system, actor_name, result):
         broadcast = handle_broadcast(value_broadcast)
         actor_invocation_response.workflow.broadcast.CopyFrom(broadcast)
 
+    if result.get_effects():
+        effects = list(map(lambda ef: to_effect(ef), [
+                       ef for ef in result.get_effects()]))
+
+        actor_invocation_response.workflow.effects.extend(effects)
+
+    if result.get_forward() != None:
+        f = result.get_forward()
+        forward = Forward()
+        forward.actor = f.actor
+        forward.action_name = f.action
+        actor_invocation_response.workflow.forward.CopyFrom(forward)
+
+    if result.get_pipe() != None:
+        p = result.get_pipe()
+        pipe = Pipe()
+        pipe.actor = p.actor
+        pipe.action_name = p.action
+        actor_invocation_response.workflow.pipe.CopyFrom(pipe)
+
     return actor_invocation_response
 
 
+def to_effect(ef: Effect):
+    # We initialize a ref actor here to ensure the target actor exists.
+    # This costs one more sidecar call but avoids actor not found errors
+    from spawn.eigr.functions.actors.api.reference import ActorRef
+    ref = ActorRef(SpawnClient(), ef.system, ef.actor, ef.parent)
+
+    req: InvocationRequest = InvocationRequest()
+    system = ActorSystem()
+    system.name = ref.actor_system
+
+    actor_id = ActorId()
+    actor_id.name = ref.actor_name
+    actor_id.system = ref.actor_system
+
+    actor = Actor()
+    actor.id.CopyFrom(actor_id)
+
+    req.system.CopyFrom(system)
+    req.actor.CopyFrom(actor)
+    req.action_name = ef.action
+    req.pooled = False
+    setattr(req, 'async', True)
+
+    if ef.payload != None:
+        req.value.CopyFrom(pack(ef.payload))
+
+    side_effect: SpawnEffect = SpawnEffect()
+    side_effect.request.CopyFrom(req)
+
+    return side_effect
+
+
 def handle_broadcast(value_broadcast):
     broadcast = Broadcast()
     broadcast.channel_group = value_broadcast.channel