Started to make Execution Async

actors/core/pom.xml

@@ -78,6 +78,11 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
         </plugins>
     </build>
     <dependencies>
+        <dependency>
+            <groupId>com.ea.orbit</groupId>
+            <artifactId>orbit-async</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>com.ea.orbit</groupId>
             <artifactId>orbit-commons</artifactId>

actors/extensions/mongodb/src/test/java/com/ea/orbit/actors/mongodb/test/MongodbPersistenceTest.java

@@ -71,7 +71,7 @@ public void checkReads() throws Exception
             SomeMatch someMatch = Actor.getReference(SomeMatch.class, "300");
             SomePlayer somePlayer = Actor.getReference(SomePlayer.class, "101");
             someMatch.addPlayer(somePlayer).get();
-            stage1.stop();
+            stage1.stop().join();
         }
         assertEquals(1, database.getCollection("SomeMatch").count());
         {

actors/stage/pom.xml

@@ -87,12 +87,6 @@ THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
             <version>${project.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>com.ea.orbit</groupId>
-            <artifactId>orbit-async</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-
         <dependency>
             <groupId>org.infinispan</groupId>
             <artifactId>infinispan-embedded</artifactId>

actors/stage/src/main/java/com/ea/orbit/actors/Stage.java

@@ -43,6 +43,7 @@
 import com.ea.orbit.actors.runtime.cloner.KryoCloner;
 import com.ea.orbit.annotation.Config;
 import com.ea.orbit.annotation.Wired;
+import com.ea.orbit.async.Await;
 import com.ea.orbit.concurrent.ExecutorUtils;
 import com.ea.orbit.concurrent.Task;
 import com.ea.orbit.container.Container;
@@ -110,24 +111,14 @@ public enum StageMode
 
     static
     {
-        // Initializes orbit async, but only if the application uses it.
         try
         {
-            Class.forName("com.ea.orbit.async.Async");
-            try
-            {
-                // async is present in the classpath, let's make sure await is initialized
-                Class.forName("com.ea.orbit.async.Await");
-            }
-            catch (Exception ex)
-            {
-                // this might be a problem, logging.
-                logger.error("Error initializing orbit-async", ex);
-            }
+            Await.init();
         }
         catch (Exception ex)
         {
-            // no problem, application doesn't use orbit async.
+            // this might be a problem, logging.
+            logger.error("Error initializing orbit-async", ex);
         }
     }
 

actors/stage/src/main/java/com/ea/orbit/actors/runtime/Execution.java

@@ -95,6 +95,8 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import static com.ea.orbit.async.Await.await;
+
 public class Execution implements Runtime
 {
 
@@ -441,44 +443,53 @@ public Activation(final ReferenceEntry entry, final Object instance)
         }
 
         // gets or creates the instance
-        public Object getOrCreateInstance() throws IllegalAccessException, InstantiationException, ExecutionException, InterruptedException
+        // gets or creates the instance
+        public Task<Object> getOrCreateInstance()
         {
-            if (instance == null)
+            try
             {
-                Object newInstance = classForName(entry.descriptor.concreteClassName).newInstance();
-                if (newInstance instanceof AbstractActor)
+                if (instance == null)
                 {
-                    final AbstractActor<?> actor = (AbstractActor<?>) newInstance;
-                    ActorTaskContext.current().setActor(actor);
-                    actor.reference = entry.reference;
+                    Object newInstance = classForName(entry.descriptor.concreteClassName).newInstance();
+                    if (newInstance instanceof AbstractActor)
+                    {
+                        final AbstractActor<?> actor = (AbstractActor<?>) newInstance;
+                        ActorTaskContext.current().setActor(actor);
+                        actor.reference = entry.reference;
 
-                    actor.stateExtension = getStorageExtensionFor(actor);
+                        actor.stateExtension = getStorageExtensionFor(actor);
 
-                    Task.allOf(getAllExtensions(LifetimeExtension.class).stream().map(v -> v.preActivation(actor))).join();
+                        await(Task.allOf(getAllExtensions(LifetimeExtension.class).stream().map(v -> v.preActivation(actor))));
 
-                    if (actor.stateExtension != null)
-                    {
-                        try
+                        if (actor.stateExtension != null)
                         {
-                            actor.readState().join();
-                        }
-                        catch (Exception ex)
-                        {
-                            if (logger.isErrorEnabled())
+                            try
                             {
-                                logger.error("Error reading actor state for: " + entry.reference, ex);
+                                await(actor.readState());
+                            }
+                            catch (Exception ex)
+                            {
+                                if (logger.isErrorEnabled())
+                                {
+                                    logger.error("Error reading actor state for: " + entry.reference, ex);
+                                }
+                                throw ex;
                             }
-                            throw ex;
                         }
+                        instance = newInstance;
+
+                        await(actor.activateAsync());
+                        await(Task.allOf(getAllExtensions(LifetimeExtension.class).stream().map(v -> v.postActivation(actor))));
                     }
-                    instance = newInstance;
 
-                    actor.activateAsync().join();
-                    Task.allOf(getAllExtensions(LifetimeExtension.class).stream().map(v -> v.postActivation(actor))).join();
                 }
 
+                return Task.fromValue(instance);
+            }
+            catch(Exception e)
+            {
+                throw new UncheckedException(e);
             }
-            return instance;
         }
     }
 
@@ -534,30 +545,30 @@ public Task<?> stop()
     {
         // * refuse new actor activations
         state = NodeCapabilities.NodeState.STOPPING;
-        hosting.notifyStateChange().join();
+        await(hosting.notifyStateChange());
 
         // * deactivate all actors
-        activationCleanup().join();
+        await(activationCleanup());
 
         // * finalize all timers
         timer.cancel();
 
         // * give extensions a chance to send a message
-        Task.allOf(extensions.stream().map(StageLifecycleListener::onPreStop)).join();
+        await(Task.allOf(extensions.stream().map(StageLifecycleListener::onPreStop)));
 
         // * stop processing new received messages (responses still work)
         // * notify rest of the cluster (no more observer messages)
         state = NodeCapabilities.NodeState.STOPPED;
-        hosting.notifyStateChange().join();
+        await(hosting.notifyStateChange());
 
         // * wait pending tasks execution
         executionSerializer.shutdown();
 
         // * cancel all pending messages, and prevents sending new ones
-        messaging.stop();
+        await(messaging.stop());
 
         // ** stop all extensions
-        Task.allOf(extensions.stream().map(Startable::stop)).join();
+        await(Task.allOf(extensions.stream().map(Startable::stop)));
 
         return Task.done();
     }
@@ -1042,7 +1053,7 @@ private Task<?> executeMessage(
             try
             {
                 bind();
-                final Object actor = activation.getOrCreateInstance();
+                final Object actor = await(activation.getOrCreateInstance());
                 context.setActor((AbstractActor<?>) actor);
 
                 Task<?> future = descriptor.invoker.safeInvoke(actor, methodId, params);

actors/test/actor-tests/src/test/java/com/ea/orbit/actors/test/FailureTest.java

@@ -64,8 +64,8 @@ public void nodeDropTest() throws ExecutionException, InterruptedException
         stage3.bind();
         assertEquals(uuid, someActor_r3.getUniqueActivationId().join());
 
-        stage1.stop();
-        stage2.stop();
+        stage1.stop().join();
+        stage2.stop().join();
 
         // a new Activation must have been created since the initial nodes where stopped.
         stage3.bind();
@@ -75,8 +75,8 @@ public void nodeDropTest() throws ExecutionException, InterruptedException
         stage3.bind();
         assertEquals(secondUUID, someActor_r4.getUniqueActivationId().join());
         // BTW, timing issues will sometimes make this fail by timeout with the real network.
-        stage3.stop();
-        stage4.stop();
+        stage3.stop().join();
+        stage4.stop().join();
 
     }
 

container/src/main/java/com/ea/orbit/container/Container.java

@@ -30,6 +30,7 @@
 
 import com.ea.orbit.annotation.Config;
 import com.ea.orbit.annotation.Order;
+import com.ea.orbit.async.Await;
 import com.ea.orbit.concurrent.Task;
 import com.ea.orbit.configuration.ContainerConfig;
 import com.ea.orbit.configuration.ContainerConfigImpl;
@@ -81,24 +82,14 @@ public class Container
 
     static
     {
-        // Initializes orbit async, but only if the application uses it.
         try
         {
-            Class.forName("com.ea.orbit.async.Async");
-            try
-            {
-                // async is present in the classpath, let's make sure await is initialized
-                Class.forName("com.ea.orbit.async.Await");
-            }
-            catch (Exception ex)
-            {
-                // this might be a problem, logging.
-                logger.error("Error initializing orbit-async", ex);
-            }
+            Await.init();
         }
         catch (Exception ex)
         {
-            // no problem, application doesn't use orbit async.
+            // this might be a problem, logging.
+            logger.error("Error initializing orbit-async", ex);
         }
     }