Feat. Initial client api

Author: sleipnir

example/spawn_example.py

@@ -2,14 +2,26 @@
 Copyright 2022 Eigr.
 Licensed under the Apache License, Version 2.0.
 """
+from spawn.eigr.functions.actors.api.reference import ActorRef
 from spawn.eigr.functions.actors.api.sdk import Spawn
 from example.joe import actor as joe_actor
-from example.domain.domain_pb2 import Reply, Request
+from example.unnamed_actor import abstract
+from example.domain.domain_pb2 import Request
 
 if __name__ == "__main__":
     request = Request()
     request.language = "erlang"
 
     spawn = Spawn()
     spawn.port(8091).proxy_port(9003).actor_system(
-        "spawn-system").add_actor(joe_actor).start()
+        "spawn-system").add_actor(joe_actor).add_actor(abstract).start()
+
+    # Get abstract actor reference called mike
+    mike_actor: ActorRef = Spawn.create_actor_ref(
+        system="spawn-system",
+        actor_name="mike",
+        parent="abs_actor",
+        state_revision=1
+    )
+
+    # mike_actor.invoke(request)

example/unnamed_actor.py

@@ -0,0 +1,19 @@
+"""
+Copyright 2022 Eigr.
+Licensed under the Apache License, Version 2.0.
+"""
+from domain.domain_pb2 import Request, Reply
+from spawn.eigr.functions.actors.api.actor import Actor
+from spawn.eigr.functions.actors.api.settings import ActorSettings, Kind
+from spawn.eigr.functions.actors.api.context import Context
+from spawn.eigr.functions.actors.api.value import Value
+
+abstract = Actor(settings=ActorSettings(
+    name="abs_actor", stateful=True, kind=Kind.UNNAMED))
+
+
+@abstract.action("setLanguage")
+def set_language(request: Request, ctx: Context) -> Value:
+    reply = Reply()
+    reply.response = "erlang"
+    return Value().of(reply, ctx.state).reply()

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

@@ -0,0 +1,56 @@
+
+from spawn.eigr.functions.actors.internal.client import SpawnClient
+
+from spawn.eigr.functions.protocol.actors.actor_pb2 import (
+    Actor,
+    ActorId,
+    ActorState,
+    Metadata,
+    ActorSettings,
+    Action,
+    FixedTimerAction,
+    ActorSnapshotStrategy,
+    ActorDeactivationStrategy,
+    ActorSystem,
+    Kind,
+    Registry,
+    TimeoutStrategy,
+)
+
+from spawn.eigr.functions.protocol.actors.protocol_pb2 import (
+    SpawnRequest,
+    SpawnResponse,
+    InvocationRequest,
+    InvocationResponse,
+    RegistrationRequest,
+    RegistrationResponse
+)
+
+
+def spawn(client: SpawnClient, system: str, actor: str, parent: str, revision: int = None):
+    actor_id = ActorId()
+    actor_id.name = actor
+    actor_id.system = system
+
+    if parent != None:
+        actor_id.parent = parent
+
+    request: SpawnRequest = SpawnRequest()
+    request.actors.append(actor_id)
+    spawn_response: SpawnResponse = client.spawn(system, request, revision)
+    return spawn_response
+
+
+class ActorRef:
+    def __init__(self, client: SpawnClient, system: str, actor: str, parent: str = None, revision: int = None):
+        self.__spawn_client = client
+        self.actor_system = system
+        self.actor_name = actor
+        self.actor_parent = parent
+        self.revision = revision
+        if parent != None:
+            spawn(self.__spawn_client, self.actor_system,
+                  self.actor_name, self.actor_parent, self.revision)
+
+    def invoke(self, request: any):
+        pass

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

@@ -8,6 +8,8 @@
 from typing import MutableMapping
 
 from spawn.eigr.functions.actors.api.actor import Actor
+from spawn.eigr.functions.actors.api.reference import ActorRef
+from spawn.eigr.functions.actors.internal.client import SpawnClient
 from spawn.eigr.functions.actors.internal.controller import ActorController
 
 from google.protobuf.any_pb2 import Any as ProtoAny
@@ -24,6 +26,7 @@ def create_app(controller: ActorController):
 
     @app.route('/api/v1/actors/actions', methods=["POST"])
     def action():
+        print(request)
         data = request.data
 
         actor_invocation_response = controller.handle_invoke(data)
@@ -54,31 +57,33 @@ class Spawn:
     __actor_entities: MutableMapping[str,
                                      Actor] = field(default_factory=dict)
 
-    # @staticmethod
-    # def invoke(name: str, command: str, arg: Any, output_type: Any) -> Any:
-    #     actorController = ActorController(
-    #         os.environ.get("PROXY_HOST", "localhost"),
-    #         os.environ.get("PROXY_PORT", "9002"),
-    #     )
-    #     actorController.invoke(name, command, arg, output_type)
+    @staticmethod
+    def create_actor_ref(system: str, actor_name: str, parent: str = None, state_revision: int = None) -> ActorRef:
+        client = SpawnClient()
+        return ActorRef(client, system, actor_name, parent, state_revision)
 
     def host(self, address: str):
         """Set the Network Host address."""
         self.__host = address
+        os.environ["USER_FUNCTION_HOST"] = address
         return self
 
     def port(self, port: int):
         """Set the Network Port address."""
         self.__port = port
+        os.environ["USER_FUNCTION_PORT"] = str(port)
         return self
 
     def proxy_host(self, host: str):
         """Set the Spawn Proxy Host Address"""
         self.__proxy_host = host
+        os.environ["PROXY_HTTP_HOST"] = host
         return self
 
     def proxy_port(self, port: int):
-        self.__proxy_port = str(port)
+        port_str: str = str(port)
+        self.__proxy_port = port_str
+        os.environ["PROXY_HTTP_PORT"] = port_str
         return self
 
     def actor_system(self, system: str = None):
@@ -93,13 +98,16 @@ def add_actor(self, actor: Actor):
 
     def start(self):
         """Start the user function and HTTP Server."""
+        import time
         if not self.__system:
             raise Exception(
                 "ActorSystem cannot be None. Use actor_system function to set an ActorSystem")
 
         address = "{}:{}".format(self.__host, self.__port)
+        client = SpawnClient()
+
         self.__controller = ActorController(
-            self.__proxy_host, self.__proxy_port, self.__system, self.__actor_entities)
+            client, self.__system, self.__actor_entities)
 
         self.__app = create_app(controller=self.__controller)
 
@@ -111,8 +119,8 @@ def start(self):
         try:
             server.start()
             client.start()
-            server.join()
-            client.join()
+            # temporary
+            time.sleep(2)
         except IOError as e:
             logging.error("Error on start Spawn %s", e.__cause__)
 

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

@@ -0,0 +1,92 @@
+
+import requests
+import os
+
+from requests.adapters import HTTPAdapter, Retry
+
+from spawn.eigr.functions.protocol.actors.protocol_pb2 import (
+    SpawnRequest,
+    SpawnResponse,
+    InvocationRequest,
+    InvocationResponse,
+    RegistrationRequest,
+    RegistrationResponse
+)
+
+_DEFAULT_HEADERS = {
+    "Accept": "application/octet-stream",
+    "Content-Type": "application/octet-stream",
+}
+_DEFAULT_MAX_RETRIES = 100
+_DEFAULT_MAX_RETRIES_BACKOFF_FACTOR = 0.2
+
+req = requests.Session()
+retries = Retry(total=_DEFAULT_MAX_RETRIES,
+                backoff_factor=_DEFAULT_MAX_RETRIES_BACKOFF_FACTOR)
+req.mount('http://', HTTPAdapter(max_retries=retries))
+
+
+class SpawnClient:
+    _instance = None
+
+    def __init__(self, host: str = None, port: str = None):
+        self.host = os.environ.get(
+            "PROXY_HTTP_HOST", "0.0.0.0") if not host else host
+        self.port = os.environ.get(
+            "PROXY_HTTP_PORT", "9001") if not port else port
+
+    def __new__(cls, *args, **kwargs):
+        if not isinstance(cls._instance, cls):
+            cls._instance = object.__new__(cls)
+
+        return cls._instance
+
+    def register(self, request: RegistrationRequest) -> RegistrationResponse:
+        proxy_url = "http://{}:{}{}".format(self.host,
+                                            self.port, "/api/v1/system")
+
+        binary_payload = request.SerializeToString()
+
+        resp = req.post(
+            proxy_url, data=binary_payload, headers=_DEFAULT_HEADERS
+        )
+
+        databytes = bytes(resp.content)
+        response = RegistrationResponse()
+        response.ParseFromString(databytes)
+        return response
+
+    def spawn(self, system: str, request: SpawnRequest, revision: int = None) -> SpawnResponse:
+        proxy_url: str = None
+        if not revision:
+            proxy_url = "http://{}:{}/api/v1/system/{}/actors/spawn".format(self.host,
+                                                                            self.port, system)
+        else:
+            proxy_url = "http://{}:{}/api/v1/system/{}/actors/spawn?revision={}".format(self.host,
+                                                                                        self.port, system, revision)
+
+        binary_payload = request.SerializeToString()
+
+        resp = req.post(
+            proxy_url, data=binary_payload, headers=_DEFAULT_HEADERS
+        )
+
+        databytes = bytes(resp.content)
+        response = SpawnResponse()
+        response.ParseFromString(databytes)
+        return response
+
+    def invoke(self, system: str, actor: str, request: InvocationRequest) -> InvocationResponse:
+        proxy_url = "http://{}:{}/api/v1/system/{}/actors/{}/invoke".format(self.host,
+                                                                            self.port, system, actor)
+
+        binary_payload = request.SerializeToString()
+
+        resp = req.post(
+            proxy_url, data=binary_payload, headers=_DEFAULT_HEADERS
+        )
+
+        databytes = bytes(resp.content)
+        response = InvocationResponse()
+        response.ParseFromString(databytes)
+        return response

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

@@ -5,13 +5,13 @@
 """
 import logging
 import platform
-import requests
 
 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.protocol.actors.actor_pb2 import (
     Actor,
@@ -42,26 +42,12 @@
 from google.protobuf import symbol_database as _symbol_database
 from google.protobuf.any_pb2 import Any as AnyProto
 
-from requests.adapters import HTTPAdapter, Retry
-
 from typing import MutableMapping
 
 _sym_db = _symbol_database.Default()
 
-_DEFAULT_HEADERS = {
-    "Accept": "application/octet-stream",
-    "Content-Type": "application/octet-stream",
-}
-_DEFAULT_MAX_RETRIES = 100
-_DEFAULT_MAX_RETRIES_BACKOFF_FACTOR = 0.2
-_REGISTER_URI = "/api/v1/system"
 TYPE_URL_PREFIX = 'type.googleapis.com/'
 
-req = requests.Session()
-retries = Retry(total=_DEFAULT_MAX_RETRIES,
-                backoff_factor=_DEFAULT_MAX_RETRIES_BACKOFF_FACTOR)
-req.mount('http://', HTTPAdapter(max_retries=retries))
-
 
 def get_payload(input):
     input_type: str = input.type_url
@@ -126,9 +112,8 @@ def handle_broadcast(value_broadcast):
 class ActorController:
     _instance = None
 
-    def __init__(self, host: str, port: str, system: str, actors: MutableMapping[str, ActorEntity]):
-        self.host = host
-        self.port = port
+    def __init__(self, client: SpawnClient, system: str, actors: MutableMapping[str, ActorEntity]):
+        self.client = client
         self.system = system
         self.actors = actors
 
@@ -180,19 +165,16 @@ def handle_invoke(self, data) -> ActorInvocationResponse:
         return handle_response(actor_system, actor_name, result)
 
     def register(self):
-        logging.info("Registering Actors on the Proxy %s", self.actors)
+        logging.info("Registering Actors on the Proxy")
         try:
 
-            proxy_url = "http://{}:{}{}".format(self.host,
-                                                self.port, _REGISTER_URI)
-
             registry = Registry()
             actor_system = ActorSystem()
             actor_system.name = self.system
 
             for actor_name, entity in self.actors.items():
-                logging.info("Registering Actor %s with Config %s",
-                             actor_name, entity)
+                logging.debug("Registering Actor %s with Config %s",
+                              actor_name, entity)
 
                 # Create actor params via ActorEntity
                 actor_template = Actor()
@@ -289,12 +271,7 @@ def register(self):
             request.service_info.CopyFrom(service_info)
             request.actor_system.CopyFrom(actor_system)
 
-            binary_payload = request.SerializeToString()
-
-            resp = req.post(
-                proxy_url, data=binary_payload, headers=_DEFAULT_HEADERS
-            )
-
+            resp = self.client.register(request)
             logging.info("Actors register response %s", resp)
         except Exception as e:
             logging.error("ERROR: %s", e)