Wrap layer run() to ensure we fetch ActorFuture results

batik/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '0.0.3'
+__version__ = '0.0.4'
 
 from .env import env
 from .layer import Layer, Actor

batik/layer.py

@@ -1,9 +1,26 @@
+from dask.distributed import ActorFuture
+import logging
 
 
 class Layer:
-    def __init__(self, client=None, **kwargs): 
-        super().__init__(**kwargs)
+    def __init__(self, layer_args={}, **kwargs): 
+        self.layer_args = layer_args
+    def _run(self, payload):
+        if isinstance(payload, ActorFuture):
+            return self.run(payload.result())
+        else:
+            return self.run(payload)
+
+
 
 class Actor:
-    def __init__(self, **kwargs): 
-        super().__init__(**kwargs)
+    def __init__(self, layer_args={}, **kwargs): 
+        self.layer_args = layer_args
+    def _run(self, payload):
+        try: 
+            if isinstance(payload, ActorFuture):
+                return self.run(payload.result())
+            else:
+                return self.run(payload)
+        except Exception as e:
+            logging.exception('')