Basic result handler for actions

Author: sleipnir

example/joe.py

@@ -16,19 +16,19 @@
 actor = Actor(settings=ActorSettings(name="joe", stateful=True))
 
 
-@actor.action("setLanguage")
-def set_language(request: Request, ctx: Context) -> Value:
+@actor.timer_action(every=1000)
+def hi(ctx: Context) -> Value:
+    print("Context {}".format(ctx))
+    new_state = JoeState()
+    new_state.languages.append("portuguese")
     return Value()\
         .of("test")\
-        .broadcast(Broadcast())\
-        .effect(Effect())\
-        .metada(Metadata())\
-        .state({})\
+        .state(new_state)\
         .reply()
 
 
-@actor.timer_action(every=1000)
-def hi(request: Request, ctx: Context) -> Value:
+@actor.action("setLanguage")
+def set_language(request: Request, ctx: Context) -> Value:
     return Value()\
         .of("test")\
         .broadcast(Broadcast())\

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

@@ -63,6 +63,7 @@ def register_timer_action_handler(function):
 
 
 def invoke(function, parameters):
+    print("Parameters {}".format(parameters))
     ordered_parameters = []
     for parameter_definition in inspect.signature(function).parameters.values():
         annotation = parameter_definition.annotation
@@ -85,7 +86,12 @@ class ActorHandler:
     entity: Actor
 
     def handle_action(self, action_name, input, ctx: Context):
-        if action_name not in self.entity.action_handlers:
-            raise Exception("Missing action handler function for Actor {} and action {}".format(
-                self.entity.settings.name, action_name))
-        return invoke(self.entity.action_handlers[action_name], [input, ctx])
+        if action_name in self.entity.action_handlers:
+            return invoke(self.entity.action_handlers[action_name], [input, ctx])
+        elif action_name in self.entity.timer_action_handlers:
+            action = self.entity.timer_action_handlers[action_name].action
+            return invoke(action, [input, ctx])
+        else:
+            error = "Missing action handler function for Actor [{}] and Action [{}]".format(
+                self.entity.settings.name, action_name)
+            raise Exception(error)

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

@@ -1,4 +1,21 @@
 
+from dataclasses import dataclass, field
+from typing import MutableMapping
+
+
+@dataclass
 class Metadata:
-    def __init__(self):
-        pass
+    __map: MutableMapping[str, str] = field(default_factory=dict)
+    __tags: MutableMapping[str, str] = field(default_factory=dict)
+
+    def put_metadata(self, key: str, value: str):
+        self.__map[key] = value
+
+    def put_tag(self, key: str, value: str):
+        self.__tags[key] = value
+
+    def get_metadata(self):
+        return self.__map
+
+    def get_tags(self):
+        return self.__tags

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

@@ -18,8 +18,8 @@ def __init__(
             stateful: bool = True,
             state_type: Generic[T] = None,
             channel: str = None,
-            deactivate_timeout: int = 30,
-            snapshot_timeout: int = 10):
+            deactivate_timeout: int = 30000,
+            snapshot_timeout: int = 2000):
         self.name = name
         self.kind = kind
         self.stateful = stateful

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

@@ -1,6 +1,7 @@
 # pylint: disable=too-few-public-methods
 # pylint: disable=arguments-differ
-from abc import ABCMeta, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
 
 from spawn.eigr.functions.actors.api.metadata import Metadata
 from spawn.eigr.functions.actors.api.workflows.broadcast import Broadcast
@@ -9,84 +10,86 @@
 from spawn.eigr.functions.actors.api.workflows.pipe import Pipe
 
 
-class IValue(metaclass=ABCMeta):
-    "The Builder Interface"
+class ReplyKind(str, Enum):
+    REPLY = 'REPLY'
+    NO_REPLY = 'NO_REPLY'
 
-    @staticmethod
-    @abstractmethod
-    def of(response, state=None):
-        "Build value from response and state"
 
-    @staticmethod
-    @abstractmethod
-    def state(state):
-        "Build value only from state"
+@dataclass
+class Value():
+    "The Concrete Builder."
+    __state = None
+    __response = None
+    __metadata: Metadata = None
+    __broadcast: Broadcast = None
+    __effect: Effect = None
+    __forward: Forward = None
+    __pipe: Pipe = None
+    __reply_kind: ReplyKind = ReplyKind.REPLY
 
-    @staticmethod
-    @abstractmethod
-    def metadata(metadata: Metadata):
-        "Build value with metadata"
+    def get_state(self):
+        return self.__state
 
-    @staticmethod
-    @abstractmethod
-    def value(response):
-        "Build value from response"
+    def get_response(self):
+        return self.__response
 
-    @staticmethod
-    @abstractmethod
-    def broadcast(broadcast: Broadcast):
-        "Create Broadcast"
+    def get_metadata(self):
+        return self.__metadata
 
-    @staticmethod
-    @abstractmethod
-    def effect(effect: Effect):
-        "Create Effect"
+    def get_broadcast(self):
+        return self.__broadcast
 
-    @staticmethod
-    @abstractmethod
-    def forward(forward: Forward):
-        "Create Forward"
+    def get_broadcast(self):
+        return self.__broadcast
 
-    @staticmethod
-    @abstractmethod
-    def pipe(pipe: Pipe):
-        "Create Pipe"
+    def get_effect(self):
+        return self.__effect
 
-    @staticmethod
-    @abstractmethod
-    def reply():
-        "Create Value response"
+    def get_forward(self):
+        return self.__forward
 
+    def get_pipe(self):
+        return self.__pipe
 
-class Value(IValue):
-    "The Concrete Builder."
+    def get_reply_kind(self):
+        return self.__reply_kind
 
-    def __init__(self):
-        pass
+    def of(self, value, state=None):
+        self.__response = value
+        self.__state = state
+        return self
 
-    def of(response, state=None):
-        pass
+    def state(self, state):
+        self.__state = state
+        return self
 
-    def state(state):
-        pass
+    def metadata(self, metadata: Metadata):
+        self.__metadata = metadata
+        return self
 
-    def metadata(metadata: Metadata):
-        pass
+    def value(self, value):
+        self.__response = value
+        return self
 
-    def value(response):
-        pass
+    def broadcast(self, broadcast: Broadcast):
+        self.__broadcast = broadcast
+        return self
 
-    def broadcast(broadcast: Broadcast):
-        pass
+    def effect(self, effect: Effect):
+        self.__effect = effect
+        return self
 
-    def effect(effect: Effect):
-        pass
+    def forward(self, forward: Forward):
+        self.__forward = forward
+        return self
 
-    def forward(forward: Forward):
-        pass
+    def pipe(self, pipe: Pipe):
+        self.__pipe = pipe
+        return self
 
-    def pipe(pipe: Pipe):
-        pass
+    def reply(self):
+        return self
 
-    def reply():
-        pass
+    def noreply(self):
+        self.__reply_kind = ReplyKind.NO_REPLY
+        return self

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

@@ -1,4 +1,4 @@
 
 class Effect:
     def __init__(self):
-        pass
+        self

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

@@ -6,8 +6,9 @@
 
 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.api.actor import TimerFunction
 
 from spawn.eigr.functions.protocol.actors.actor_pb2 import (
     Actor,
@@ -29,24 +30,75 @@
     ActorInvocation,
     ActorInvocationResponse,
     Context,
+    Noop,
     RegistrationRequest,
     RegistrationResponse,
     ServiceInfo,
 )
 
+from google.protobuf import symbol_database as _symbol_database
+from google.protobuf.any_pb2 import Any as AnyProto
+
 import logging
 import platform
 import requests
 
 from typing import Any, MutableMapping
 
+_sym_db = _symbol_database.Default()
+
 _DEFAULT_HEADERS = {
     "Accept": "application/octet-stream",
     "Content-Type": "application/octet-stream",
 }
 
 _REGISTER_URI = "/api/v1/system"
 
+TYPE_URL_PREFIX = 'type.googleapis.com/'
+
+
+def get_payload(input):
+    input_type: str = input.payload.type_url
+    if input_type.startswith(TYPE_URL_PREFIX):
+        input_type = input_type[len(TYPE_URL_PREFIX):]
+    input_class = _sym_db.GetSymbol(input_type)
+    input = input_class()
+    input.ParseFromString(input.payload.value)
+    return input
+
+
+def pack(input):
+    any = AnyProto()
+    any.Pack(input)
+    return any
+
+
+def handle_response(system, actor_name, result):
+    print("Result ----- {}".format(result))
+    actor_invocation_response = ActorInvocationResponse()
+    actor_invocation_response.actor_name = actor_name
+    actor_invocation_response.actor_system = system
+
+    updated_context = Context()
+
+    if result.get_metadata() != None and len(result.get_metadata().get_metadata()) > 0:
+        print("Metadata -----------".format(result.metadata))
+        updated_context.metadata = result.get_metadata.get_metadata()
+
+    if result.get_metadata() != None and len(result.get_metadata().get_tags()) > 0:
+        updated_context.tags = result.get_metadata().get_tags()
+
+    updated_context.state.CopyFrom(pack(result.get_state()))
+
+    actor_invocation_response.updated_context.CopyFrom(updated_context)
+
+    if result.get_reply_kind() == ReplyKind.NO_REPLY:
+        actor_invocation_response.noop = Noop()
+    elif result.get_reply_kind == ReplyKind.REPLY:
+        actor_invocation_response.value = pack(result.get_response())
+
+    return actor_invocation_response
+
 
 class ActorController:
     _instance = None
@@ -81,17 +133,23 @@ def handle_invoke(self, data) -> ActorInvocationResponse:
         entity = self.actors[actor_name] if not actor_parent else self.actors[actor_parent]
 
         handler = ActorHandler(entity)
+        current_context = actor_invocation.current_context
+
+        input = None if actor_invocation.WhichOneof(
+            "payload") == "noop" else get_payload(actor_invocation.value)
+
+        ctx = ActorContext(state=None, caller=actor_invocation.caller.name,
+                           metadata=current_context.metadata, tags=current_context.tags)
 
-        # Update Context
-        updated_context = Context()
+        result = handler.handle_action(
+            action_name=actor_invocation.action_name, input=input, ctx=ctx)
 
-        # Then send ActorInvocationResponse back to the caller
-        actor_invocation_response = ActorInvocationResponse()
-        actor_invocation_response.actor_name = actor_name
-        actor_invocation_response.actor_system = actor_system
-        actor_invocation_response.updated_context.CopyFrom(updated_context)
+        if not isinstance(result, Value):
+            raise Exception(
+                "Action did not return a valid type in its response. Valid Value found {}".format(type(result)))
 
-        return actor_invocation_response
+        # Handle result value
+        return handle_response(actor_system, actor_name, result)
 
     def register(self):
         logging.info("Registering Actors on the Proxy %s", self.actors)