Fixes for RPC shadow services (see below)

- RPC return value now properly passed on to remote callback

- Shadow services are now being used if the message bus is
offline, remote communication is disabled or no remote
endpoint for the service exists

- No longer attempt to contact remote endpoint if message
was deferred or was already delivered to a shadow service

- No longer swallow exceptions thrown by shadow service
implementations

Author: csadilek

errai-bus/src/main/java/org/jboss/errai/bus/client/api/base/MessageBuilder.java

@@ -18,14 +18,14 @@
 
 import org.jboss.errai.bus.client.api.BusErrorCallback;
 import org.jboss.errai.bus.client.api.HasEncoded;
-import org.jboss.errai.bus.client.api.messaging.Message;
+import org.jboss.errai.bus.client.api.RoutingFlag;
 import org.jboss.errai.bus.client.api.builder.DefaultRemoteCallBuilder;
 import org.jboss.errai.bus.client.api.builder.MessageBuildCommand;
 import org.jboss.errai.bus.client.api.builder.MessageBuildSendableWithReply;
 import org.jboss.errai.bus.client.api.builder.MessageBuildSubject;
 import org.jboss.errai.bus.client.api.builder.MessageReplySendable;
+import org.jboss.errai.bus.client.api.messaging.Message;
 import org.jboss.errai.bus.client.api.messaging.MessageProvider;
-import org.jboss.errai.bus.client.api.RoutingFlag;
 import org.jboss.errai.common.client.api.RemoteCallback;
 
 /**
@@ -119,7 +119,6 @@ public static MessageBuildCommand<MessageBuildSendableWithReply> createMessage(f
    * @return a <tt>MessageBuildSubject</tt> which essentially is a <tt>Message</tt>, but ensures that the user
    *         constructs messages properly
    */
-  @SuppressWarnings({ "unchecked" })
   public static MessageBuildSubject<MessageReplySendable> createConversation(final Message message) {
     final Message newMessage = provider.get();
     newMessage.setFlag(RoutingFlag.NonGlobalRouting);
@@ -143,7 +142,6 @@ public static MessageBuildSubject<MessageReplySendable> createConversation(final
    * @return a <tt>MessageBuildSubject</tt> which essentially is a <tt>Message</tt>, but ensures that the user
    *         constructs messages properly
    */
-  @SuppressWarnings({ "unchecked" })
   public static MessageBuildCommand<MessageReplySendable> createConversation(final Message message, final String subject) {
     final Message newMessage = provider.get();
     if (newMessage instanceof HasEncoded) {

errai-bus/src/main/java/org/jboss/errai/bus/client/api/builder/DefaultRemoteCallBuilder.java

@@ -173,7 +173,7 @@ public RemoteCallResponseDef endpoint(String endPointName, Annotation[] qualifie
 
         if (qualifiers != null) {
           final List<String> qualNames = new ArrayList<String>(qualifiers.length);
-          for (Annotation a : qualifiers) {
+          for (final Annotation a : qualifiers) {
             qualNames.add(a.annotationType().getName());
           }
 

errai-bus/src/main/java/org/jboss/errai/bus/client/framework/ClientMessageBusImpl.java

@@ -121,12 +121,12 @@ public void callback(final Message message) {
       final String errorTo = message.get(String.class, MessageParts.ErrorTo);
 
       if (errorTo == null || DefaultErrorCallback.CLIENT_ERROR_SUBJECT.equals(errorTo)) {
-        Throwable t = message.get(Throwable.class, MessageParts.Throwable);
+        final Throwable t = message.get(Throwable.class, MessageParts.Throwable);
         Optional.ofNullable(GWT.getUncaughtExceptionHandler())
                 .ifPresent(h -> h.onUncaughtException(t));
 
         if (!uncaughtExceptionHandlers.isEmpty()) {
-          for (UncaughtExceptionHandler handler : uncaughtExceptionHandlers) {
+          for (final UncaughtExceptionHandler handler : uncaughtExceptionHandlers) {
             handler.onUncaughtException(t);
           }
         }
@@ -319,12 +319,12 @@ private void setBusToInitializableState() {
 
   private void setupDefaultHandlers() {
     if (availableHandlers != null) {
-      for (TransportHandler handler : availableHandlers.values()) {
+      for (final TransportHandler handler : availableHandlers.values()) {
         handler.close();
       }
     }
 
-    Map<String, TransportHandler> m = new LinkedHashMap<String, TransportHandler>();
+    final Map<String, TransportHandler> m = new LinkedHashMap<String, TransportHandler>();
     m.put(Capabilities.WebSockets.name(), new WebsocketHandler(ClientMessageBusImpl.this));
     m.put(Capabilities.SSE.name(), new SSEHandler(ClientMessageBusImpl.this));
     m.put(Capabilities.LongPolling.name(),
@@ -673,7 +673,7 @@ public void callback(final Message message) {
         try {
           callback.callback(message);
         }
-        catch (Exception e) {
+        catch (final Exception e) {
           handleCallbackError(message, e);
         }
       }
@@ -779,29 +779,26 @@ public void send(final Message message) {
     logger.debug("send({})", message.getParts());
 
     try {
-      boolean deferred = false;
+      boolean delivered = false;
       final boolean localOnly = message.isFlagSet(RoutingFlag.DeliverLocalOnly);
       final String subject = message.getSubject();
 
       if (message.hasPart(MessageParts.ToSubject)) {
         if (isRemoteCommunicationEnabled() && !localOnly) {
           if (getState().isShadowDeliverable() && shadowSubscriptions.containsKey(subject)) {
             deliverToSubscriptions(shadowSubscriptions, subject, message);
-            deferred = true;
+            delivered = true;
           }
           else if (getState() != BusState.CONNECTED) {
             logger.debug("deferred: {}", message);
             deferredMessages.add(message);
-            deferred = true;
+            delivered = true;
+          }
+          else if (remotes.containsKey(subject)) {
+            logger.debug("sent to remote: {}", message);
+            remotes.get(subject).callback(message);
+            delivered = true;
           }
-        }
-
-        boolean routedToRemote = false;
-
-        if (!localOnly && remotes.containsKey(subject)) {
-          logger.debug("sent to remote: {}", message);
-          remotes.get(subject).callback(message);
-          routedToRemote = true;
         }
 
         if (subscriptions.containsKey(subject)) {
@@ -810,16 +807,21 @@ else if (getState() != BusState.CONNECTED) {
         else if (localSubscriptions.containsKey(subject)) {
           deliverToSubscriptions(localSubscriptions, subject, message);
         }
-        else if (!deferred && !routedToRemote) {
-          throw new NoSubscribersToDeliverTo(subject);
+        else if (!delivered) {
+          if (shadowSubscriptions.containsKey(subject)) {
+            deliverToSubscriptions(shadowSubscriptions, subject, message);
+          }
+          else {
+            throw new NoSubscribersToDeliverTo(subject);
+          }
         }
       }
       else {
         throw new RuntimeException("Cannot send message using this method"
             + " if the message does not contain a ToSubject field.");
       }
     }
-    catch (RuntimeException e) {
+    catch (final RuntimeException e) {
       callErrorHandler(message, e);
     }
   }
@@ -986,7 +988,7 @@ private void handleCallbackError(final Message message, final Throwable t) {
       try {
         defaultErrorHandling = message.getErrorCallback().error(message, t);
       }
-      catch (Throwable secondaryError) {
+      catch (final Throwable secondaryError) {
         logger.error("Encountered an error while calling error callback for message to " + message.getSubject(), secondaryError);
       }
     }
@@ -997,7 +999,7 @@ private void handleCallbackError(final Message message, final Throwable t) {
   }
 
   private void loadRpcProxies() {
-    RpcProxyLoader proxyLoader = ((RpcProxyLoader) GWT.create(RpcProxyLoader.class));
+    final RpcProxyLoader proxyLoader = ((RpcProxyLoader) GWT.create(RpcProxyLoader.class));
     proxyLoader.loadProxies(ClientMessageBusImpl.this);
   }
 
@@ -1218,7 +1220,7 @@ else if (newState == BusState.LOCAL_ONLY) {
         try {
           et.deliverTo(lifecycleListeners.get(i), e);
         }
-        catch (Throwable t) {
+        catch (final Throwable t) {
           logger.error("listener threw exception: " + t);
           t.printStackTrace();
         }

errai-bus/src/main/java/org/jboss/errai/bus/client/util/BusToolsCli.java

@@ -115,7 +115,7 @@ private static Map<String, Object> decodePayloadMap(final EJValue value) {
   }
 
   private static final QueueSession clientSession = new QueueSession() {
-    private Map<String, Object> attributes = new HashMap<String, Object>();
+    private final Map<String, Object> attributes = new HashMap<String, Object>();
 
     @Override
     public String getSessionId() {
@@ -194,7 +194,6 @@ public static void setAutoDemarshall(final boolean autoDemarshall1) {
    * @param path
    *     path to use when sending requests to the JAX-RS endpoint
    */
-  @SuppressWarnings("UnusedDeclaration")
   public static native void setApplicationRoot(final String path) /*-{
       if (path == null) {
           $wnd.erraiBusApplicationRoot = undefined;

errai-ioc-bus-support/src/main/java/org/jboss/errai/ioc/support/bus/rebind/ShadowServiceDecorator.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright (C) 2012 Red Hat, Inc. and/or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.jboss.errai.ioc.support.bus.rebind;
+
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.List;
+
+import org.jboss.errai.bus.client.ErraiBus;
+import org.jboss.errai.bus.client.api.ClientMessageBus;
+import org.jboss.errai.bus.client.api.base.MessageBuilder;
+import org.jboss.errai.bus.client.api.messaging.Message;
+import org.jboss.errai.bus.client.api.messaging.MessageCallback;
+import org.jboss.errai.bus.server.annotations.Remote;
+import org.jboss.errai.bus.server.annotations.ShadowService;
+import org.jboss.errai.codegen.Parameter;
+import org.jboss.errai.codegen.Statement;
+import org.jboss.errai.codegen.VariableReference;
+import org.jboss.errai.codegen.builder.AnonymousClassStructureBuilder;
+import org.jboss.errai.codegen.builder.BlockBuilder;
+import org.jboss.errai.codegen.builder.ElseBlockBuilder;
+import org.jboss.errai.codegen.builder.impl.ObjectBuilder;
+import org.jboss.errai.codegen.util.EmptyStatement;
+import org.jboss.errai.codegen.util.If;
+import org.jboss.errai.codegen.util.ProxyUtil;
+import org.jboss.errai.codegen.util.Refs;
+import org.jboss.errai.codegen.util.Stmt;
+import org.jboss.errai.ioc.client.api.CodeDecorator;
+import org.jboss.errai.ioc.rebind.ioc.extension.IOCDecoratorExtension;
+import org.jboss.errai.ioc.rebind.ioc.injector.api.Decorable;
+import org.jboss.errai.ioc.rebind.ioc.injector.api.FactoryController;
+
+/**
+ * Generates logic to register client-side shadow services for Errai's message
+ * bus. Shadow services are used when:
+ * <ul>
+ * <li>Remote communication is turned off
+ * <li>Errai's message bus is not in connected state
+ * <li>A remote endpoint for the service doesn't exist
+ * </ul>
+ * 
+ * @author Mike Brock
+ * @author Christian Sadilek <csadilek@redhat.com>
+ */
+@CodeDecorator
+public class ShadowServiceDecorator extends IOCDecoratorExtension<ShadowService> {
+  public ShadowServiceDecorator(Class<ShadowService> decoratesWith) {
+    super(decoratesWith);
+  }
+
+  @Override
+  public void generateDecorator(final Decorable decorable, final FactoryController controller) {
+    final ShadowService shadowService = (ShadowService) decorable.getAnnotation();
+    String serviceName = null;
+
+    Statement subscribeShadowStatement = null;
+    final Class<?> javaClass = decorable.getType().asClass();
+    for (final Class<?> intf : javaClass.getInterfaces()) {
+      if (intf.isAnnotationPresent(Remote.class)) {
+        serviceName = intf.getName() + ":RPC";
+
+        final AnonymousClassStructureBuilder builder = generateMethodDelegates(decorable.getAccessStatement(), intf,
+                controller);
+
+        subscribeShadowStatement = Stmt.castTo(ClientMessageBus.class, Stmt.invokeStatic(ErraiBus.class, "get"))
+                .invoke("subscribeShadow", serviceName, builder.finish());
+      }
+
+      if (serviceName == null) {
+        if (shadowService.value().equals("")) {
+          serviceName = decorable.getName();
+        }
+        else {
+          serviceName = shadowService.value();
+        }
+
+        subscribeShadowStatement = Stmt.castTo(ClientMessageBus.class, Stmt.invokeStatic(ErraiBus.class, "get"))
+                .invoke("subscribeShadow", serviceName, decorable.getAccessStatement());
+      }
+
+      controller.addFactoryInitializationStatements(Collections.singletonList(subscribeShadowStatement));
+    }
+  }
+
+  private AnonymousClassStructureBuilder generateMethodDelegates(final Statement accessStatement, final Class<?> intf,
+          final FactoryController controller) {
+
+    final BlockBuilder<AnonymousClassStructureBuilder> builder = ObjectBuilder.newInstanceOf(MessageCallback.class)
+            .extend().publicOverridesMethod("callback", Parameter.of(Message.class, "message"))
+            .append(Stmt.declareVariable("commandType", String.class,
+                    Stmt.loadVariable("message").invoke("getCommandType")))
+            .append(Stmt.declareVariable("methodParms", List.class,
+                    Stmt.loadVariable("message").invoke("get", List.class, Stmt.loadLiteral("MethodParms"))));
+
+    for (final Method method : intf.getMethods()) {
+      if (ProxyUtil.isMethodInInterface(intf, method)) {
+        final Class<?>[] parameterTypes = method.getParameterTypes();
+        final VariableReference[] objects = new VariableReference[parameterTypes.length];
+        final BlockBuilder<ElseBlockBuilder> blockBuilder = If
+                .cond(Stmt.loadLiteral(ProxyUtil.createCallSignature(intf, method)).invoke("equals",
+                        Stmt.loadVariable("commandType")));
+        
+        for (int i = 0; i < parameterTypes.length; i++) {
+          final Class<?> parameterType = parameterTypes[i];
+          blockBuilder.append(Stmt.declareVariable("var" + i, parameterType,
+                  Stmt.castTo(parameterType, Stmt.loadVariable("methodParms").invoke("get", i))));
+          objects[i] = Refs.get("var" + i);
+        }
+
+        final boolean hasReturnType = !method.getReturnType().equals(void.class);
+        final Statement methodInvocation = Stmt.nestedCall(controller.contextGetInstanceStmt()).invoke(method.getName(), (Object[]) objects);
+        blockBuilder.append(Stmt.try_()
+                .append((hasReturnType) ? Stmt.declareFinalVariable("ret", method.getReturnType(), methodInvocation) : methodInvocation)
+                .append((hasReturnType) ? Stmt.invokeStatic(MessageBuilder.class, "createConversation", Stmt.loadVariable("message"))
+                        .invoke("subjectProvided").invoke("with", "MethodReply", Refs.get("ret"))
+                        .invoke("noErrorHandling").invoke("sendNowWith", Stmt.invokeStatic(ErraiBus.class, "get"))
+                        : EmptyStatement.INSTANCE)
+                .finish().catch_(Throwable.class, "throwable")
+                .append(Stmt.throw_(RuntimeException.class, Stmt.loadVariable("throwable"))).finish());
+        builder.append(blockBuilder.finish());
+      }
+    }
+    return builder.finish();
+  }
+}