ISPN-6610 Stop using forkInvocationSync

Interceptors must use onReturn() for things they need
to do after command.perform(). While forkInvocationSync()
still works, it doesn't allow asynchronous execution.

* Fix forkInvocation-related problems in BaseSequentialInvocationContext.
* Make BaseSequentialInvocationContext.invokeInterceptors invoke
  visitCommand in sequence for all interceptors, to allow inlining.
* Move visit notification from CallInterceptor to EntryWrappingInterceptor

cdi/embedded/src/test/java/org/infinispan/cdi/embedded/test/event/CacheEventTest.java

@@ -38,8 +38,7 @@
 @Test(groups = "functional", testName = "cdi.test.event.CacheEventTest")
 public class CacheEventTest extends Arquillian {
 
-   private final NonTxInvocationContext invocationContext = new NonTxInvocationContext(null, AnyEquivalence.getInstance(),
-                                                                                       null);
+   private final NonTxInvocationContext invocationContext = new NonTxInvocationContext(null, AnyEquivalence.getInstance());
 
    @Inject
    @Cache1

core/src/main/java/org/infinispan/container/EntryFactoryImpl.java

@@ -1,7 +1,5 @@
 package org.infinispan.container;
 
-import org.infinispan.distribution.DistributionManager;
-import org.infinispan.metadata.Metadata;
 import org.infinispan.atomic.Delta;
 import org.infinispan.atomic.DeltaAware;
 import org.infinispan.configuration.cache.Configuration;
@@ -13,8 +11,10 @@
 import org.infinispan.container.entries.RepeatableReadEntry;
 import org.infinispan.container.entries.StateChangingEntry;
 import org.infinispan.context.InvocationContext;
+import org.infinispan.distribution.DistributionManager;
 import org.infinispan.factories.annotations.Inject;
 import org.infinispan.factories.annotations.Start;
+import org.infinispan.metadata.Metadata;
 import org.infinispan.util.TimeService;
 import org.infinispan.util.concurrent.IsolationLevel;
 import org.infinispan.util.logging.Log;
@@ -31,7 +31,7 @@
 public class EntryFactoryImpl implements EntryFactory {
 
    private static final Log log = LogFactory.getLog(EntryFactoryImpl.class);
-   private final boolean trace = log.isTraceEnabled();
+   private final static boolean trace = log.isTraceEnabled();
 
    private boolean useRepeatableRead;
    private DataContainer container;

core/src/main/java/org/infinispan/context/NonTransactionalInvocationContextFactory.java

@@ -35,7 +35,7 @@ public InvocationContext createInvocationContext(boolean isWrite, int keyCount)
       if (keyCount == 1) {
          return new SingleKeyNonTxInvocationContext(null, keyEq);
       } else if (keyCount > 0) {
-         return new NonTxInvocationContext(keyCount, null, keyEq, interceptorChain);
+         return new NonTxInvocationContext(keyCount, null, keyEq);
       }
       return createInvocationContext(null, false);
    }
@@ -47,8 +47,7 @@ public InvocationContext createInvocationContext(Transaction tx, boolean implici
 
    @Override
    public NonTxInvocationContext createNonTxInvocationContext() {
-      NonTxInvocationContext ctx = new NonTxInvocationContext(null, keyEq, interceptorChain);
-      return ctx;
+      return new NonTxInvocationContext(null, keyEq);
    }
 
    @Override
@@ -58,8 +57,7 @@ public InvocationContext createSingleKeyNonTxInvocationContext() {
 
    @Override
    public NonTxInvocationContext createRemoteInvocationContext(Address origin) {
-      NonTxInvocationContext ctx = new NonTxInvocationContext(origin, keyEq, interceptorChain);
-      return ctx;
+      return new NonTxInvocationContext(origin, keyEq);
    }
 
    @Override

core/src/main/java/org/infinispan/context/SequentialInvocationContext.java

@@ -72,7 +72,7 @@ public interface SequentialInvocationContext {
     * synchronously.
     *
     * <p>Usually it is easier to use than
-    * {@link #forkInvocation(VisitableCommand, SequentialInterceptor.ReturnHandler)}.
+    * {@link #forkInvocation(VisitableCommand, SequentialInterceptor.ForkReturnHandler)}.
     * However, it is not recommended, because any asynchronous work in the remaining interceptors will block
     * the calling thread.</p>
     *

core/src/main/java/org/infinispan/context/TransactionalInvocationContextFactory.java

@@ -68,18 +68,18 @@ public InvocationContext createInvocationContext(Transaction tx, boolean implici
          throw new IllegalArgumentException("Cannot create a transactional context without a valid Transaction instance.");
       }
       LocalTransaction localTransaction = transactionTable.getOrCreateLocalTransaction(tx, implicitTransaction);
-      return new LocalTxInvocationContext(localTransaction, interceptorChain);
+      return new LocalTxInvocationContext(localTransaction);
    }
 
    @Override
    public LocalTxInvocationContext createTxInvocationContext(LocalTransaction localTransaction) {
-      return new LocalTxInvocationContext(localTransaction, interceptorChain);
+      return new LocalTxInvocationContext(localTransaction);
    }
 
    @Override
    public RemoteTxInvocationContext createRemoteTxInvocationContext(
          RemoteTransaction tx, Address origin) {
-      RemoteTxInvocationContext ctx = new RemoteTxInvocationContext(tx, interceptorChain);
+      RemoteTxInvocationContext ctx = new RemoteTxInvocationContext(tx);
       return ctx;
    }
 
@@ -104,7 +104,7 @@ private Transaction getRunningTx() {
    }
 
    protected final NonTxInvocationContext newNonTxInvocationContext(Address origin) {
-      NonTxInvocationContext ctx = new NonTxInvocationContext(origin, keyEq, interceptorChain);
+      NonTxInvocationContext ctx = new NonTxInvocationContext(origin, keyEq);
       return ctx;
    }
 }

core/src/main/java/org/infinispan/context/impl/AbstractTxInvocationContext.java

@@ -2,7 +2,6 @@
 
 import org.infinispan.commands.write.WriteCommand;
 import org.infinispan.container.entries.CacheEntry;
-import org.infinispan.interceptors.SequentialInterceptorChain;
 import org.infinispan.remoting.transport.Address;
 import org.infinispan.transaction.impl.AbstractCacheTransaction;
 import org.infinispan.transaction.xa.GlobalTransaction;
@@ -25,8 +24,7 @@ public abstract class AbstractTxInvocationContext<T extends AbstractCacheTransac
 
    private final T cacheTransaction;
 
-   protected AbstractTxInvocationContext(T cacheTransaction, Address origin,
-                                         SequentialInterceptorChain interceptorChain) {
+   protected AbstractTxInvocationContext(T cacheTransaction, Address origin) {
       super(origin);
       if (cacheTransaction == null) {
          throw new NullPointerException("CacheTransaction cannot be null");

core/src/main/java/org/infinispan/context/impl/ImmutableContext.java

@@ -5,7 +5,6 @@
 import org.infinispan.container.entries.CacheEntry;
 import org.infinispan.context.InvocationContext;
 import org.infinispan.interceptors.SequentialInterceptor;
-import org.infinispan.interceptors.impl.BaseSequentialInvocationContext;
 import org.infinispan.remoting.transport.Address;
 
 import java.util.Collections;
@@ -19,7 +18,7 @@
  * 
  * @author Sanne Grinovero <sanne@infinispan.org> (C) 2011 Red Hat Inc.
  */
-public final class ImmutableContext extends BaseSequentialInvocationContext {
+public final class ImmutableContext implements InvocationContext {
 
    public static final ImmutableContext INSTANCE = new ImmutableContext();
 

core/src/main/java/org/infinispan/context/impl/LocalTxInvocationContext.java

@@ -1,6 +1,5 @@
 package org.infinispan.context.impl;
 
-import org.infinispan.interceptors.SequentialInterceptorChain;
 import org.infinispan.remoting.transport.Address;
 import org.infinispan.transaction.impl.LocalTransaction;
 
@@ -20,9 +19,8 @@
  */
 public class LocalTxInvocationContext extends AbstractTxInvocationContext<LocalTransaction> {
 
-   public LocalTxInvocationContext(LocalTransaction localTransaction,
-                                   SequentialInterceptorChain interceptorChain) {
-      super(localTransaction, null, interceptorChain);
+   public LocalTxInvocationContext(LocalTransaction localTransaction) {
+      super(localTransaction, null);
    }
 
    @Override

core/src/main/java/org/infinispan/context/impl/NonTxInvocationContext.java

@@ -4,7 +4,6 @@
 import org.infinispan.commons.equivalence.Equivalence;
 import org.infinispan.commons.util.CollectionFactory;
 import org.infinispan.container.entries.CacheEntry;
-import org.infinispan.interceptors.SequentialInterceptorChain;
 import org.infinispan.remoting.transport.Address;
 
 import java.util.Collections;
@@ -27,15 +26,13 @@ public class NonTxInvocationContext extends AbstractInvocationContext {
    private Object lockOwner;
 
 
-   public NonTxInvocationContext(int numEntries, Address origin, Equivalence<Object> keyEq,
-                                 SequentialInterceptorChain interceptorChain) {
+   public NonTxInvocationContext(int numEntries, Address origin, Equivalence<Object> keyEq) {
       super(origin);
       lookedUpEntries = CollectionFactory.makeMap(CollectionFactory.computeCapacity(numEntries), keyEq, AnyEquivalence.getInstance());
       this.keyEq = keyEq;
    }
 
-   public NonTxInvocationContext(Address origin, Equivalence<Object> keyEq,
-                                 SequentialInterceptorChain interceptorChain) {
+   public NonTxInvocationContext(Address origin, Equivalence<Object> keyEq) {
       super(origin);
       lookedUpEntries = CollectionFactory.makeMap(INITIAL_CAPACITY, keyEq, AnyEquivalence.getInstance());
       this.keyEq = keyEq;

core/src/main/java/org/infinispan/context/impl/RemoteTxInvocationContext.java

@@ -1,6 +1,5 @@
 package org.infinispan.context.impl;
 
-import org.infinispan.interceptors.SequentialInterceptorChain;
 import org.infinispan.transaction.impl.RemoteTransaction;
 
 import javax.transaction.Transaction;
@@ -15,9 +14,8 @@
  */
 public class RemoteTxInvocationContext extends AbstractTxInvocationContext<RemoteTransaction> {
 
-   public RemoteTxInvocationContext(RemoteTransaction cacheTransaction,
-                                    SequentialInterceptorChain interceptorChain) {
-      super(cacheTransaction, cacheTransaction.getGlobalTransaction().getAddress(), interceptorChain);
+   public RemoteTxInvocationContext(RemoteTransaction cacheTransaction) {
+      super(cacheTransaction, cacheTransaction.getGlobalTransaction().getAddress());
    }
 
    @Override

core/src/main/java/org/infinispan/factories/AbstractComponentRegistry.java

@@ -817,7 +817,11 @@ public void buildInjectionMethodsList() throws ClassNotFoundException {
                }
                Method m = meta.getMethod();
                if (m == null) {
-                  m = ReflectionUtil.findMethod(clazz, meta.getMethodName(), parameterClasses);
+                  try {
+                     m = ReflectionUtil.findMethod(clazz, meta.getMethodName(), parameterClasses);
+                  } catch (CacheException e) {
+                     throw new CacheException("Injection method not found in class " + clazz + ": " + meta.getMethodName() + Arrays.toString(parameterClasses), e);
+                  }
                   meta.setMethod(m);
                }
             }
@@ -848,9 +852,8 @@ public boolean equals(Object o) {
          PrioritizedMethod that = (PrioritizedMethod) o;
 
          if (component != null ? !component.equals(that.component) : that.component != null) return false;
-         if (metadata != null ? !metadata.equals(that.metadata) : that.metadata != null) return false;
+         return metadata != null ? metadata.equals(that.metadata) : that.metadata == null;
 
-         return true;
       }
 
       @Override

core/src/main/java/org/infinispan/interceptors/SequentialInterceptor.java

@@ -29,7 +29,7 @@ public interface SequentialInterceptor {
     * <li>The interceptor can call {@link SequentialInvocationContext#shortCircuit(Object)} in order to skip
     * the execution of the rest of the chain (and the command itself).</li>
     * <li>The interceptor can call
-    * {@link SequentialInvocationContext#forkInvocation(VisitableCommand, ReturnHandler)} in order to invoke
+    * {@link SequentialInvocationContext#forkInvocation(VisitableCommand, ForkReturnHandler)} in order to invoke
     * a new command, starting with the next interceptor in the chain. The return handler then behaves as
     * another {@code visitCommand} invocation: it can allow the invocation of the original command to continue
     * with the next interceptor, short-circuit the invocation, or fork another command.</li>
@@ -39,8 +39,9 @@ public interface SequentialInterceptor {
     * </ul>
     *
     * <p>Thread safety: The interceptor must only invoke methods on the context or command in the thread
-    * calling {@link #visitCommand(InvocationContext, VisitableCommand)} or
-    * {@link ReturnHandler#handle(InvocationContext, VisitableCommand, Object, Throwable)}.</p>
+    * calling {@code visitCommand(InvocationContext, VisitableCommand)},
+    * {@link ReturnHandler#handle(InvocationContext, VisitableCommand, Object, Throwable)}, or
+    * {@link ForkReturnHandler#handle(InvocationContext, VisitableCommand, Object, Throwable)}.</p>
     */
    CompletableFuture<Void> visitCommand(InvocationContext ctx, VisitableCommand command) throws Throwable;
 
@@ -53,20 +54,23 @@ public interface SequentialInterceptor {
     * {@code CompletableFuture}. Returning {@code null} is allowed, and will continue the invocation with
     * the next return handler, without modifying the return value and without creating a new
     * {@code CompletableFuture} instance.</p>
+    *
+    * TODO Have different interfaces for processing the return value and for finally-like handlers that don't have access to the return value
+    * TODO At least prohibit null return values and require an already-completed CF with a known value instead.
     */
    interface ReturnHandler {
-      CompletableFuture<Object> handle(InvocationContext ctx, VisitableCommand command, Object rv,
+      CompletableFuture<Object> handle(InvocationContext rCtx, VisitableCommand rCommand, Object rv,
             Throwable throwable) throws Throwable;
    }
 
-   /*
-   * A return handler installed with
-   * {@link SequentialInvocationContext#forkInvocation(VisitableCommand, ForkHandler)}.
+   /**
+    * A return handler installed with
+    * {@link SequentialInvocationContext#forkInvocation(VisitableCommand, ForkReturnHandler)}.
     *
     * <p>It must behave just like {@link #visitCommand(InvocationContext, VisitableCommand)}.</p>
-   */
+    */
    interface ForkReturnHandler {
-      CompletableFuture<Void> handle(InvocationContext ctx, VisitableCommand command, Object rv,
+      CompletableFuture<Void> handle(InvocationContext rCtx, VisitableCommand rCommand, Object rv,
             Throwable throwable) throws Throwable;
    }
 }

core/src/main/java/org/infinispan/interceptors/SequentialInterceptorChain.java

@@ -3,6 +3,7 @@
 import org.infinispan.commands.VisitableCommand;
 import org.infinispan.commons.util.Experimental;
 import org.infinispan.context.InvocationContext;
+import org.infinispan.context.SequentialInvocationContext;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -17,21 +18,24 @@
  */
 @Experimental
 public interface SequentialInterceptorChain {
+   /**
+    * @return An immutable list of the current interceptors.
+    */
    List<SequentialInterceptor> getInterceptors();
 
    /**
-    * Inserts the given interceptor at the specified position in the chain (o based indexing).
+    * Inserts the given interceptor at the specified position in the chain (0 based indexing).
     *
-    * @throws IllegalArgumentException if the position is invalid (e.g. 5 and there are only 2 interceptors in the
-    *                                  chain)
+    * @throws IllegalArgumentException if the position is invalid (e.g. 5 and there are only 2 interceptors
+    *       in the chain)
     */
    void addInterceptor(SequentialInterceptor interceptor, int position);
 
    /**
-    * Removes the interceptor at the given postion.
+    * Removes the interceptor at the given position.
     *
-    * @throws IllegalArgumentException if the position is invalid (e.g. 5 and there are only 2 interceptors in the
-    *                                  chain)
+    * @throws IllegalArgumentException if the position is invalid (e.g. 5 and there are only 2 interceptors
+    *       in the chain)
     */
    void removeInterceptor(int position);
 
@@ -41,28 +45,29 @@ public interface SequentialInterceptorChain {
    int size();
 
    /**
-    * Removes all the occurences of supplied interceptor type from the chain.
+    * Removes all the occurrences of supplied interceptor type from the chain.
     */
    void removeInterceptor(Class<? extends SequentialInterceptor> clazz);
 
    /**
     * Adds a new interceptor in list after an interceptor of a given type.
     *
-    * @return true if the interceptor was added; i.e. the afterInterceptor exists
+    * @return true if the interceptor was added; i.e. the {@code afterInterceptor} exists
     */
    boolean addInterceptorAfter(SequentialInterceptor toAdd, Class<? extends
          SequentialInterceptor> afterInterceptor);
 
    /**
-    * Adds a new interceptor in list after an interceptor of a given type.
+    * Adds a new interceptor in list before an interceptor of a given type.
     *
-    * @return true if the interceptor was added; i.e. the afterInterceptor exists
+    * @return true if the interceptor was added; i.e. the {@code beforeInterceptor} exists
     */
    boolean addInterceptorBefore(SequentialInterceptor toAdd,
                                                 Class<? extends SequentialInterceptor> beforeInterceptor);
 
    /**
-    * Replaces an existing interceptor of the given type in the interceptor chain with a new interceptor instance passed as parameter.
+    * Replaces an existing interceptor of the given type in the interceptor chain with a new interceptor
+    * instance passed as parameter.
     *
     * @param replacingInterceptor        the interceptor to add to the interceptor chain
     * @param toBeReplacedInterceptorType the type of interceptor that should be swapped with the new one
@@ -80,11 +85,11 @@ boolean replaceInterceptor(SequentialInterceptor replacingInterceptor,
     * Walks the command through the interceptor chain. The received ctx is being passed in.
     *
     * <p>Note: Reusing the context for multiple invocations is allowed. However, the two invocations
-    * must not overlap, so calling {@link #invoke(InvocationContext, VisitableCommand)} from an interceptor
-    * is not allowed. If an interceptor wants to invoke a new command and cannot use {@link org.infinispan
-    * .context.SequentialInvocationContext#forkInvocation(VisitableCommand, SequentialInterceptor
-    * .ForkReturnHandler)} or
-    * {@link org.infinispan.context.SequentialInvocationContext#forkInvocationSync(VisitableCommand)},
+    * must not overlap, so calling {@code invoke(InvocationContext, VisitableCommand)} from an interceptor
+    * is not allowed.
+    * If an interceptor wants to invoke a new command and cannot use
+    * {@link SequentialInvocationContext#forkInvocation(VisitableCommand, SequentialInterceptor.ForkReturnHandler)}
+    * or {@link SequentialInvocationContext#forkInvocationSync(VisitableCommand)},
     * it must first copy the invocation context with {@link InvocationContext#clone()}.</p>
     */
    Object invoke(InvocationContext ctx, VisitableCommand command);