Interceptor cleanup

hapi-fhir-base/src/main/java/ca/uhn/fhir/util/VersionUtil.java

@@ -58,7 +58,9 @@ public class VersionUtil {
 		try (InputStream is = VersionUtil.class.getResourceAsStream("/ca/uhn/fhir/hapi-fhir-base-build.properties")) {
 
 			Properties p = new Properties();
-			p.load(is);
+			if (is != null) {
+				p.load(is);
+			}
 
 			ourVersion = p.getProperty("hapifhir.version");
 			ourVersion = defaultIfBlank(ourVersion, "(unknown)");

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/config/BaseConfig.java

@@ -42,9 +42,9 @@ import javax.annotation.Nonnull;
  * 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.
@@ -57,10 +57,12 @@ import javax.annotation.Nonnull;
 @Configuration
 @EnableScheduling
 @EnableJpaRepositories(basePackages = "ca.uhn.fhir.jpa.dao.data")
-@ComponentScan(basePackages = "ca.uhn.fhir.jpa", excludeFilters={
+@ComponentScan(basePackages = "ca.uhn.fhir.jpa", excludeFilters = {
-		  @ComponentScan.Filter(type=FilterType.ASSIGNABLE_TYPE, value=BaseConfig.class),
+	@ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = BaseConfig.class),
-		  @ComponentScan.Filter(type=FilterType.ASSIGNABLE_TYPE, value=WebSocketConfigurer.class),
+	@ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, value = WebSocketConfigurer.class),
-			@ComponentScan.Filter(type=FilterType.REGEX, pattern="ca.uhn.fhir.jpa.subscription.module.standalone.*")})
+	@ComponentScan.Filter(type = FilterType.REGEX, pattern = ".*\\.test\\..*"),
+	@ComponentScan.Filter(type = FilterType.REGEX, pattern = ".*Test.*"),
+	@ComponentScan.Filter(type = FilterType.REGEX, pattern = "ca.uhn.fhir.jpa.subscription.module.standalone.*")})
 
 public abstract class BaseConfig implements SchedulingConfigurer {
 

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/interceptor/test/InterceptorRegistryTest.java

@@ -1,8 +1,10 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
+package ca.uhn.fhir.jpa.interceptor.test;
 
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.Pointcut;
+import ca.uhn.fhir.jpa.model.interceptor.executor.InterceptorRegistry;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionHook;
+import ca.uhn.fhir.jpa.model.interceptor.api.HookParams;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionInterceptor;
+import ca.uhn.fhir.jpa.model.interceptor.api.Interceptor;
+import ca.uhn.fhir.jpa.model.interceptor.api.Pointcut;
+import ca.uhn.fhir.jpa.model.interceptor.api.Hook;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -21,24 +23,25 @@ import static org.hamcrest.Matchers.contains;
 import static org.junit.Assert.*;
 
 @RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration(classes = {SubscriptionInterceptorRegistryTest.MyCtxConfig.class})
+@ContextConfiguration(classes = {InterceptorRegistryTest.InterceptorRegistryTestCtxConfig.class})
-public class SubscriptionInterceptorRegistryTest {
+public class InterceptorRegistryTest {
 
 	private static boolean ourNext_beforeRestHookDelivery_Return2;
 	private static boolean ourNext_beforeRestHookDelivery_Return1;
 	private static List<String> ourInvocations = new ArrayList<>();
 	private static CanonicalSubscription ourLastCanonicalSubscription;
-	private static ResourceDeliveryMessage ourLastResourceDeliveryMessage;
+	private static ResourceDeliveryMessage ourLastResourceDeliveryMessage0;
+	private static ResourceDeliveryMessage ourLastResourceDeliveryMessage1;
 
 	@Autowired
-	private SubscriptionInterceptorRegistry mySubscriptionInterceptorRegistry;
+	private InterceptorRegistry myInterceptorRegistry;
 
 	@Test
 	public void testGlobalInterceptorsAreFound() {
-		List<Object> globalInterceptors = mySubscriptionInterceptorRegistry.getGlobalInterceptors();
+		List<Object> globalInterceptors = myInterceptorRegistry.getGlobalInterceptorsForUnitTest();
 		assertEquals(2, globalInterceptors.size());
-		assertTrue(globalInterceptors.get(0).getClass().toString(), globalInterceptors.get(0) instanceof MyInterceptorOne);
+		assertTrue(globalInterceptors.get(0).getClass().toString(), globalInterceptors.get(0) instanceof MyTestInterceptorOne);
-		assertTrue(globalInterceptors.get(1).getClass().toString(), globalInterceptors.get(1) instanceof MyInterceptorTwo);
+		assertTrue(globalInterceptors.get(1).getClass().toString(), globalInterceptors.get(1) instanceof MyTestInterceptorTwo);
 	}
 
 	@Test
@@ -46,11 +49,12 @@ public class SubscriptionInterceptorRegistryTest {
 		ResourceDeliveryMessage msg = new ResourceDeliveryMessage();
 		CanonicalSubscription subs = new CanonicalSubscription();
 		HookParams params = new HookParams(msg, subs);
-		boolean outcome = mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+		boolean outcome = myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY, params);
 		assertTrue(outcome);
 
-		assertThat(ourInvocations, contains("MyInterceptorOne.beforeRestHookDelivery", "MyInterceptorTwo.beforeRestHookDelivery"));
+		assertThat(ourInvocations, contains("MyTestInterceptorOne.beforeRestHookDelivery", "MyTestInterceptorTwo.beforeRestHookDelivery"));
-		assertSame(msg, ourLastResourceDeliveryMessage);
+		assertSame(msg, ourLastResourceDeliveryMessage0);
+		assertNull(ourLastResourceDeliveryMessage1);
 		assertSame(subs, ourLastCanonicalSubscription);
 	}
 
@@ -61,10 +65,10 @@ public class SubscriptionInterceptorRegistryTest {
 		ResourceDeliveryMessage msg = new ResourceDeliveryMessage();
 		CanonicalSubscription subs = new CanonicalSubscription();
 		HookParams params = new HookParams(msg, subs);
-		boolean outcome = mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+		boolean outcome = myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY, params);
 		assertFalse(outcome);
 
-		assertThat(ourInvocations, contains("MyInterceptorOne.beforeRestHookDelivery"));
+		assertThat(ourInvocations, contains("MyTestInterceptorOne.beforeRestHookDelivery"));
 	}
 
 	@Test
@@ -73,10 +77,10 @@ public class SubscriptionInterceptorRegistryTest {
 		CanonicalSubscription subs = new CanonicalSubscription();
 		HookParams params = new HookParams(msg, subs);
 		try {
-			mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+			myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY, params);
 			fail();
 		} catch (AssertionError e) {
-			// good
+			assertEquals("Wrong hook parameters, wanted [CanonicalSubscription, ResourceDeliveryMessage] and found [CanonicalSubscription, Integer]", e.getMessage());
 		}
 	}
 
@@ -86,17 +90,22 @@ public class SubscriptionInterceptorRegistryTest {
 		ourNext_beforeRestHookDelivery_Return1 = true;
 		ourNext_beforeRestHookDelivery_Return2 = true;
 		ourLastCanonicalSubscription = null;
-		ourLastResourceDeliveryMessage = null;
+		ourLastResourceDeliveryMessage0 = null;
+		ourLastResourceDeliveryMessage1 = null;
 		ourInvocations.clear();
 	}
 
 	@Configuration
 	@ComponentScan(basePackages = "ca.uhn.fhir.jpa.model")
-	public static class MyCtxConfig {
+	static class InterceptorRegistryTestCtxConfig {
 
+		/**
+		 * Note: Orders are deliberately reversed to make sure we get the orders right
+		 * using the @Order annotation
+		 */
 		@Bean
-		public SubscriptionInterceptorRegistry subscriptionInterceptorRegistry() {
+		public MyTestInterceptorTwo interceptor1() {
-			return new SubscriptionInterceptorRegistry();
+			return new MyTestInterceptorTwo();
 		}
 
 		/**
@@ -104,41 +113,37 @@ public class SubscriptionInterceptorRegistryTest {
 		 * using the @Order annotation
 		 */
 		@Bean
-		public MyInterceptorTwo interceptor1() {
+		public MyTestInterceptorOne interceptor2() {
-			return new MyInterceptorTwo();
+			return new MyTestInterceptorOne();
-		}
-
-		/**
-		 * Note: Orders are deliberately reversed to make sure we get the orders right
-		 * using the @Order annotation
-		 */
-		@Bean
-		public MyInterceptorOne interceptor2() {
-			return new MyInterceptorOne();
 		}
 
 	}
 
-	@SubscriptionInterceptor
+	@Interceptor
 	@Order(100)
-	public static class MyInterceptorOne {
+	public static class MyTestInterceptorOne {
 
-		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		public MyTestInterceptorOne() {
+			super();
+		}
+
+		@Hook(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY)
 		public boolean beforeRestHookDelivery(CanonicalSubscription theCanonicalSubscription) {
 			ourLastCanonicalSubscription = theCanonicalSubscription;
-			ourInvocations.add("MyInterceptorOne.beforeRestHookDelivery");
+			ourInvocations.add("MyTestInterceptorOne.beforeRestHookDelivery");
 			return ourNext_beforeRestHookDelivery_Return1;
 		}
 
 	}
 
-	@SubscriptionInterceptor
+	@Interceptor
 	@Order(200)
-	public static class MyInterceptorTwo {
+	public static class MyTestInterceptorTwo {
-		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		@Hook(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY)
-		public boolean beforeRestHookDelivery(ResourceDeliveryMessage theResourceDeliveryMessage) {
+		public boolean beforeRestHookDelivery(ResourceDeliveryMessage theResourceDeliveryMessage0, ResourceDeliveryMessage theResourceDeliveryMessage1) {
-			ourLastResourceDeliveryMessage = theResourceDeliveryMessage;
+			ourLastResourceDeliveryMessage0 = theResourceDeliveryMessage0;
-			ourInvocations.add("MyInterceptorTwo.beforeRestHookDelivery");
+			ourLastResourceDeliveryMessage1 = theResourceDeliveryMessage1;
+			ourInvocations.add("MyTestInterceptorTwo.beforeRestHookDelivery");
 			return ourNext_beforeRestHookDelivery_Return2;
 		}
 	}

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/subscription/resthook/RestHookWithInterceptorR4Test.java

@@ -2,11 +2,11 @@ package ca.uhn.fhir.jpa.subscription.resthook;
 
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.jpa.config.StoppableSubscriptionDeliveringRestHookSubscriber;
+import ca.uhn.fhir.jpa.model.interceptor.api.Hook;
+import ca.uhn.fhir.jpa.model.interceptor.api.Interceptor;
+import ca.uhn.fhir.jpa.model.interceptor.api.Pointcut;
 import ca.uhn.fhir.jpa.subscription.BaseSubscriptionsR4Test;
 import ca.uhn.fhir.jpa.subscription.module.CanonicalSubscription;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.Pointcut;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionHook;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionInterceptor;
 import ca.uhn.fhir.jpa.subscription.module.subscriber.ResourceDeliveryMessage;
 import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.MethodOutcome;
@@ -28,7 +28,7 @@ import static org.junit.Assert.*;
 /**
  * Test the rest-hook subscriptions
  */
-@ContextConfiguration(classes = {RestHookWithInterceptorR4Test.MyCtxConfig.class})
+@ContextConfiguration(classes = {RestHookWithInterceptorR4Test.MyTestCtxConfig.class})
 public class RestHookWithInterceptorR4Test extends BaseSubscriptionsR4Test {
 
 	private static final Logger ourLog = LoggerFactory.getLogger(RestHookWithInterceptorR4Test.class);
@@ -118,11 +118,11 @@ public class RestHookWithInterceptorR4Test extends BaseSubscriptionsR4Test {
 	}
 
 	@Configuration
-	public static class MyCtxConfig {
+	static class MyTestCtxConfig {
 
 		@Bean
-		public MyInterceptor interceptor() {
+		public MyTestInterceptor interceptor() {
-			return new MyInterceptor();
+			return new MyTestInterceptor();
 		}
 
 	}
@@ -130,17 +130,17 @@ public class RestHookWithInterceptorR4Test extends BaseSubscriptionsR4Test {
 	/**
 	 * Interceptor class
 	 */
-	@SubscriptionInterceptor
+	@Interceptor
-	public static class MyInterceptor {
+	public static class MyTestInterceptor {
 
 		/**
 		 * Constructor
 		 */
-		public MyInterceptor() {
+		public MyTestInterceptor() {
 			ourLog.info("Creating interceptor");
 		}
 
-		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		@Hook(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY)
 		public boolean beforeRestHookDelivery(ResourceDeliveryMessage theDeliveryMessage, CanonicalSubscription theSubscription) {
 			if (ourNextModifyResourceId) {
 				theDeliveryMessage.getPayload(ourCtx).setId(new IdType("Observation/A"));
@@ -153,7 +153,7 @@ public class RestHookWithInterceptorR4Test extends BaseSubscriptionsR4Test {
 			return ourNextBeforeRestHookDeliveryReturn;
 		}
 
-		@SubscriptionHook(Pointcut.AFTER_REST_HOOK_DELIVERY)
+		@Hook(Pointcut.SUBSCRIPTION_AFTER_REST_HOOK_DELIVERY)
 		public boolean afterRestHookDelivery(ResourceDeliveryMessage theDeliveryMessage, CanonicalSubscription theSubscription) {
 			ourHitAfterRestHookDelivery = true;
 			return ourNextAfterRestHookDeliveryReturn;

hapi-fhir-jpaserver-model/pom.xml

@@ -97,6 +97,11 @@
 			<artifactId>jscience</artifactId>
 		</dependency>
 
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-collections4</artifactId>
+		</dependency>
+
 		<!-- Java -->
 		<dependency>
 			<groupId>javax.annotation</groupId>

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/Hook.java

@@ -1,4 +1,4 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.api;
+package ca.uhn.fhir.jpa.model.interceptor.api;
 
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
@@ -7,18 +7,18 @@ import java.lang.annotation.Target;
 
 /**
  * This annotation should be placed on
- * {@link SubscriptionInterceptor Subscription Interceptor}
+ * {@link Interceptor Subscription Interceptor}
  * bean methods.
  * <p>
- *     Methods with this annotation are invoked immediately before a REST HOOK
+ * Methods with this annotation are invoked immediately before a REST HOOK
- *     subscription delivery
+ * subscription delivery
  * </p>
  *
- * @see SubscriptionInterceptor
+ * @see Interceptor
  */
 @Target(ElementType.METHOD)
 @Retention(RetentionPolicy.RUNTIME)
-public @interface SubscriptionHook {
+public @interface Hook {
 
 	/**
 	 * Provides the specific point where this method should be invoked

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/HookParams.java

@@ -0,0 +1,56 @@
+package ca.uhn.fhir.jpa.model.interceptor.api;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class HookParams {
+
+	private ListMultimap<Class<?>, Object> myParams = ArrayListMultimap.create();
+
+	/**
+	 * Constructor
+	 */
+	public HookParams() {
+	}
+
+	/**
+	 * Constructor
+	 */
+	public HookParams(Object... theParams) {
+		for (Object next : theParams) {
+			add(next);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private <T> void add(T theNext) {
+		Class<T> nextClass = (Class<T>) theNext.getClass();
+		add(nextClass, theNext);
+	}
+
+	public <T> HookParams add(Class<T> theType, T theParam) {
+		myParams.put(theType, theParam);
+		return this;
+	}
+
+	@SuppressWarnings("unchecked")
+	public <T> T get(Class<T> theParamType, int theIndex) {
+		List<T> objects = (List<T>) myParams.get(theParamType);
+		T retVal = null;
+		if (objects.size() > theIndex) {
+			retVal = objects.get(theIndex);
+		}
+		return retVal;
+	}
+
+	/**
+	 * Multivalued parameters will be returned twice in this list
+	 */
+	public List<String> getTypesAsSimpleName() {
+		return myParams.values().stream().map(t -> t.getClass().getSimpleName()).collect(Collectors.toList());
+	}
+
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/IAnonymousLambdaHook.java

@@ -0,0 +1,16 @@
+package ca.uhn.fhir.jpa.model.interceptor.api;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * This is currently only here for unit tests!
+ *
+ * DO NOT USE IN NON-TEST CODE. Maybe this will change in the future?
+ */
+@FunctionalInterface
+@VisibleForTesting
+public interface IAnonymousLambdaHook {
+
+	void invoke(HookParams theArgs);
+
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/IInterceptorRegistry.java

@@ -0,0 +1,23 @@
+package ca.uhn.fhir.jpa.model.interceptor.api;
+
+import com.google.common.annotations.VisibleForTesting;
+
+public interface IInterceptorRegistry {
+
+	@VisibleForTesting
+	void registerAnonymousHookForUnitTest(Pointcut thePointcut, IAnonymousLambdaHook theHook);
+
+	@VisibleForTesting
+	void clearAnonymousHookForUnitTest();
+
+	/**
+	 * Invoke the interceptor methods
+	 */
+	boolean callHooks(Pointcut thePointcut, HookParams theParams);
+
+	/**
+	 * Invoke the interceptor methods
+	 */
+	boolean callHooks(Pointcut thePointcut, Object... theParams);
+
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/Interceptor.java

@@ -1,4 +1,4 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.api;
+package ca.uhn.fhir.jpa.model.interceptor.api;
 
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
@@ -10,5 +10,5 @@ import java.lang.annotation.Target;
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target(ElementType.TYPE)
-public @interface SubscriptionInterceptor {
+public @interface Interceptor {
 }

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/api/Pointcut.java

@@ -0,0 +1,63 @@
+package ca.uhn.fhir.jpa.model.interceptor.api;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Value for {@link Hook#value()}
+ */
+public enum Pointcut {
+
+	/**
+	 * Invoked immediately after the delivery of a REST HOOK subscription.
+	 * <p>
+	 * When this hook is called, all processing is complete so this hook should not
+	 * make any changes to the parameters.
+	 * </p>
+	 * Hooks may accept the following parameters:
+	 * <ul>
+	 * <li>ca.uhn.fhir.jpa.subscription.module.CanonicalSubscription</li>
+	 * <li>ca.uhn.fhir.jpa.subscription.module.subscriber.ResourceDeliveryMessage</li>
+	 * </ul>
+	 */
+	SUBSCRIPTION_AFTER_REST_HOOK_DELIVERY("CanonicalSubscription", "ResourceDeliveryMessage"),
+
+	/**
+	 * Invoked immediately before the delivery of a REST HOOK subscription.
+	 * <p>
+	 * Hooks may make changes to the delivery payload, or make changes to the
+	 * canonical subscription such as adding headers, modifying the channel
+	 * endpoint, etc.
+	 * </p>
+	 * Hooks may accept the following parameters:
+	 * <ul>
+	 * <li>ca.uhn.fhir.jpa.subscription.module.CanonicalSubscription</li>
+	 * <li>ca.uhn.fhir.jpa.subscription.module.subscriber.ResourceDeliveryMessage</li>
+	 * </ul>
+	 */
+	SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY("CanonicalSubscription", "ResourceDeliveryMessage"),
+
+	/**
+	 * Invoked whenever a persisted resource (a resource that has just been stored in the
+	 * database via a create/update/patch/etc.) has been checked for whether any subscriptions
+	 * were triggered as a result of the operation
+	 * Hooks may accept the following parameters:
+	 * <ul>
+	 * <li>ca.uhn.fhir.jpa.subscription.module.ResourceModifiedMessage</li>
+	 * </ul>
+	 */
+	SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED("ResourceModifiedMessage")
+
+	;
+
+	private final List<String> myParameterTypes;
+
+	Pointcut(String... theParameterTypes) {
+		myParameterTypes = Collections.unmodifiableList(Arrays.asList(theParameterTypes));
+	}
+
+	public List<String> getParameterTypes() {
+		return myParameterTypes;
+	}
+}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/interceptor/executor/InterceptorRegistry.java

@@ -1,10 +1,11 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
+package ca.uhn.fhir.jpa.model.interceptor.executor;
 
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.Pointcut;
+import ca.uhn.fhir.jpa.model.interceptor.api.*;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionHook;
+import com.google.common.annotations.VisibleForTesting;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.SubscriptionInterceptor;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ListMultimap;
+import com.google.common.collect.Multimaps;
+import org.apache.commons.collections4.ListUtils;
 import org.apache.commons.lang3.Validate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -18,27 +19,50 @@ import org.springframework.stereotype.Component;
 import javax.annotation.Nonnull;
 import javax.annotation.PostConstruct;
 import java.lang.reflect.Method;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Collection;
+import java.util.concurrent.atomic.AtomicInteger;
-import java.util.IdentityHashMap;
-import java.util.List;
 
 @Component
-public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptorRegistry, ApplicationContextAware {
+public class InterceptorRegistry implements IInterceptorRegistry, ApplicationContextAware {
-	private static final Logger ourLog = LoggerFactory.getLogger(SubscriptionInterceptorRegistry.class);
+	private static final Logger ourLog = LoggerFactory.getLogger(InterceptorRegistry.class);
 	private ApplicationContext myAppCtx;
 	private List<Object> myGlobalInterceptors = new ArrayList<>();
-	private ListMultimap<Pointcut, Invoker> myInvokers = ArrayListMultimap.create();
+	private ListMultimap<Pointcut, BaseInvoker> myInvokers = ArrayListMultimap.create();
+	private ListMultimap<Pointcut, BaseInvoker> myAnonymousInvokers = Multimaps.synchronizedListMultimap(ArrayListMultimap.create());
 
-	List<Object> getGlobalInterceptors() {
+	/**
+	 * Constructor
+	 */
+	public InterceptorRegistry() {
+		super();
+	}
+
+	@VisibleForTesting
+	public List<Object> getGlobalInterceptorsForUnitTest() {
 		return myGlobalInterceptors;
 	}
 
+
+	@Override
+	@VisibleForTesting
+	public void registerAnonymousHookForUnitTest(Pointcut thePointcut, IAnonymousLambdaHook theHook) {
+		Validate.notNull(thePointcut);
+		Validate.notNull(theHook);
+
+		myAnonymousInvokers.put(thePointcut, new AnonymousLambdaInvoker(theHook));
+	}
+
+	@Override
+	@VisibleForTesting
+	public void clearAnonymousHookForUnitTest() {
+		myAnonymousInvokers.clear();
+	}
+
 	@PostConstruct
 	public void start() {
 
 		// Grab the global interceptors
-		String[] globalInterceptorNames = myAppCtx.getBeanNamesForAnnotation(SubscriptionInterceptor.class);
+		String[] globalInterceptorNames = myAppCtx.getBeanNamesForAnnotation(Interceptor.class);
 		for (String nextName : globalInterceptorNames) {
 			Object nextGlobalInterceptor = myAppCtx.getBean(nextName);
 			myGlobalInterceptors.add(nextGlobalInterceptor);
@@ -50,9 +74,9 @@ public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptor
 		// Pull out the hook methods
 		for (Object nextInterceptor : myGlobalInterceptors) {
 			for (Method nextMethod : nextInterceptor.getClass().getDeclaredMethods()) {
-				SubscriptionHook hook = AnnotationUtils.findAnnotation(nextMethod, SubscriptionHook.class);
+				Hook hook = AnnotationUtils.findAnnotation(nextMethod, Hook.class);
 				if (hook != null) {
-					Invoker invoker = new Invoker(nextInterceptor, nextMethod);
+					HookInvoker invoker = new HookInvoker(hook, nextInterceptor, nextMethod);
 					for (Pointcut nextPointcut : hook.value()) {
 						myInvokers.put(nextPointcut, invoker);
 					}
@@ -84,12 +108,17 @@ public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptor
 
 	@Override
 	public boolean callHooks(Pointcut thePointcut, HookParams theParams) {
+		assert haveAppropriateParams(thePointcut, theParams);
+
+		// Anonymous hooks first
+		List<BaseInvoker> invokers = ListUtils.union(
+			myAnonymousInvokers.get(thePointcut),
+			myInvokers.get(thePointcut));
 
 		/*
 		 * Call each hook in order
 		 */
-		List<Invoker> invokers = myInvokers.get(thePointcut);
+		for (BaseInvoker nextInvoker : invokers) {
-		for (Invoker nextInvoker : invokers) {
 			boolean shouldContinue = nextInvoker.invoke(theParams);
 			if (!shouldContinue) {
 				return false;
@@ -99,22 +128,55 @@ public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptor
 		return true;
 	}
 
+	/**
+	 * Only call this when assertions are enabled, it's expensive
+	 */
+	private boolean haveAppropriateParams(Pointcut thePointcut, HookParams theParams) {
+		List<String> givenTypes = theParams.getTypesAsSimpleName();
+		List<String> wantedTypes = new ArrayList<>(thePointcut.getParameterTypes());
+		givenTypes.sort(Comparator.naturalOrder());
+		wantedTypes.sort(Comparator.naturalOrder());
+		if (!givenTypes.equals(wantedTypes)) {
+			throw new AssertionError("Wrong hook parameters, wanted " + wantedTypes + " and found " + givenTypes);
+		}
+		return true;
+	}
+
 	@Override
 	public boolean callHooks(Pointcut thePointcut, Object... theParams) {
 		return callHooks(thePointcut, new HookParams(theParams));
 	}
 
-	private class Invoker {
+	private abstract class BaseInvoker {
+		abstract boolean invoke(HookParams theParams);
+	}
+
+	private class AnonymousLambdaInvoker extends BaseInvoker {
+		private final IAnonymousLambdaHook myHook;
+
+		public AnonymousLambdaInvoker(IAnonymousLambdaHook theHook) {
+			myHook = theHook;
+		}
+
+		@Override
+		boolean invoke(HookParams theParams) {
+			myHook.invoke(theParams);
+			return true;
+		}
+	}
+
+	private class HookInvoker extends BaseInvoker {
 
 		private final Object myInterceptor;
 		private final boolean myReturnsBoolean;
 		private final Method myMethod;
 		private final Class<?>[] myParameterTypes;
+		private final int[] myParameterIndexes;
 
 		/**
 		 * Constructor
 		 */
-		private Invoker(@Nonnull Object theInterceptor, @Nonnull Method theHookMethod) {
+		private HookInvoker(Hook theHook, @Nonnull Object theInterceptor, @Nonnull Method theHookMethod) {
 			myInterceptor = theInterceptor;
 			myParameterTypes = theHookMethod.getParameterTypes();
 			myMethod = theHookMethod;
@@ -126,13 +188,26 @@ public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptor
 				Validate.isTrue(Void.class.equals(returnType), "Method does not return boolean or void: %s", theHookMethod);
 				myReturnsBoolean = false;
 			}
+
+			myParameterIndexes = new int[myParameterTypes.length];
+			Map<Class<?>, AtomicInteger> typeToCount = new HashMap<>();
+			for (int i = 0; i < myParameterTypes.length; i++) {
+				AtomicInteger counter = typeToCount.computeIfAbsent(myParameterTypes[i], t -> new AtomicInteger(0));
+				myParameterIndexes[i] = counter.getAndIncrement();
+			}
 		}
 
+		/**
+		 * @return Returns true/false if the hook method returns a boolean, returns true otherwise
+		 */
+		@Override
 		boolean invoke(HookParams theParams) {
+
 			Object[] args = new Object[myParameterTypes.length];
 			for (int i = 0; i < myParameterTypes.length; i++) {
 				Class<?> nextParamType = myParameterTypes[i];
-				Object nextParamValue = theParams.get(nextParamType);
+				int nextParamIndex = myParameterIndexes[i];
+				Object nextParamValue = theParams.get(nextParamType, nextParamIndex);
 				args[i] = nextParamValue;
 			}
 
@@ -153,7 +228,4 @@ public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptor
 
 	}
 
-	private static <T> boolean equals(Collection<T> theLhs, Collection<T> theRhs) {
-		return theLhs.size() == theRhs.size() && theLhs.containsAll(theRhs) && theRhs.containsAll(theLhs);
-	}
 }

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/subscription/interceptor/api/Pointcut.java

@@ -1,28 +0,0 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.api;
-
-/**
- * Value for {@link SubscriptionHook#value()}
- */
-public enum Pointcut {
-
-	/**
-	 * Invoked immediately after the delivery of a REST HOOK subscription.
-	 * <p>
-	 * When this hook is called, all processing is complete so this hook should not
-	 * make any changes to the parameters.
-	 * </p>
-	 */
-	AFTER_REST_HOOK_DELIVERY,
-
-	/**
-	 * Invoked immediately before the delivery of a REST HOOK subscription.
-	 * <p>
-	 * Hooks may make changes to the delivery payload, or make changes to the
-	 * canonical subscription such as adding headers, modifying the channel
-	 * endpoint, etc.
-	 * </p>
-	 */
-	BEFORE_REST_HOOK_DELIVERY;
-
-
-}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/subscription/interceptor/executor/HookParams.java

@@ -1,48 +0,0 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
-
-import org.apache.commons.lang3.Validate;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Set;
-
-public class HookParams {
-
-	private Map<Class<?>, Object> myParams = new HashMap<>();
-
-	/**
-	 * Constructor
-	 */
-	public HookParams() {
-	}
-
-	/**
-	 * Constructor
-	 */
-	public HookParams(Object... theParams) {
-		for (Object next : theParams) {
-			add(next);
-		}
-	}
-
-	@SuppressWarnings("unchecked")
-	private <T> void add(T theNext) {
-		Class<T> nextClass = (Class<T>) theNext.getClass();
-		add(nextClass, theNext);
-	}
-
-	public <T> HookParams add(Class<T> theType, T theParam) {
-		Validate.isTrue(myParams.containsKey(theType) == false, "Already have param of type %s", theType);
-		myParams.put(theType, theParam);
-		return this;
-	}
-
-	@SuppressWarnings("unchecked")
-	public <T> T get(Class<T> theParamType) {
-		return (T) myParams.get(theParamType);
-	}
-
-	Set<Class<?>> getTypes() {
-		return myParams.keySet();
-	}
-}

hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/subscription/interceptor/executor/ISubscriptionInterceptorRegistry.java

@@ -1,17 +0,0 @@
-package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
-
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.Pointcut;
-
-public interface ISubscriptionInterceptorRegistry {
-
-	/**
-	 * Invoke the interceptor methods
-	 */
-	boolean callHooks(Pointcut thePointcut, HookParams theParams);
-
-	/**
-	 * Invoke the interceptor methods
-	 */
-	boolean callHooks(Pointcut thePointcut, Object... theParams);
-
-}

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/module/ResourceModifiedMessage.java

@@ -9,9 +9,9 @@ package ca.uhn.fhir.jpa.subscription.module;
  * 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.
@@ -52,11 +52,15 @@ public class ResourceModifiedMessage implements IResourceMessage {
 	@JsonIgnore
 	private transient IBaseResource myPayloadDecoded;
 
-	// For JSON
+	/**
+	 * Constructor
+	 */
 	public ResourceModifiedMessage() {
+		super();
 	}
 
 	public ResourceModifiedMessage(FhirContext theFhirContext, IBaseResource theResource, OperationTypeEnum theOperationType) {
+		this();
 		setId(theResource.getIdElement());
 		setOperationType(theOperationType);
 		if (theOperationType != OperationTypeEnum.DELETE) {

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/module/subscriber/SubscriptionDeliveringRestHookSubscriber.java

@@ -21,8 +21,8 @@ package ca.uhn.fhir.jpa.subscription.module.subscriber;
  */
 
 import ca.uhn.fhir.jpa.subscription.module.CanonicalSubscription;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.api.Pointcut;
+import ca.uhn.fhir.jpa.model.interceptor.api.Pointcut;
-import ca.uhn.fhir.jpa.model.subscription.interceptor.executor.ISubscriptionInterceptorRegistry;
+import ca.uhn.fhir.jpa.model.interceptor.api.IInterceptorRegistry;
 import ca.uhn.fhir.rest.api.EncodingEnum;
 import ca.uhn.fhir.rest.api.RequestTypeEnum;
 import ca.uhn.fhir.rest.client.api.*;
@@ -52,7 +52,7 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 	IResourceRetriever myResourceRetriever;
 	private Logger ourLog = LoggerFactory.getLogger(SubscriptionDeliveringRestHookSubscriber.class);
 	@Autowired
-	private ISubscriptionInterceptorRegistry mySubscriptionInterceptorRegistry;
+	private IInterceptorRegistry myInterceptorRegistry;
 
 	protected void deliverPayload(ResourceDeliveryMessage theMsg, CanonicalSubscription theSubscription, EncodingEnum thePayloadType, IGenericClient theClient) {
 		IBaseResource payloadResource = getAndMassagePayload(theMsg, theSubscription);
@@ -132,8 +132,8 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 	public void handleMessage(ResourceDeliveryMessage theMessage) throws MessagingException {
 		CanonicalSubscription subscription = theMessage.getSubscription();
 
-		// Interceptor call: BEFORE_REST_HOOK_DELIVERY
+		// Interceptor call: SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY
-		if (!mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, theMessage, subscription)) {
+		if (!myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY, theMessage, subscription)) {
 			return;
 		}
 
@@ -168,8 +168,8 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 
 		deliverPayload(theMessage, subscription, payloadType, client);
 
-		// Interceptor call: AFTER_REST_HOOK_DELIVERY
+		// Interceptor call: SUBSCRIPTION_AFTER_REST_HOOK_DELIVERY
-		if (!mySubscriptionInterceptorRegistry.callHooks(Pointcut.AFTER_REST_HOOK_DELIVERY, theMessage, subscription)) {
+		if (!myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_AFTER_REST_HOOK_DELIVERY, theMessage, subscription)) {
 			//noinspection UnnecessaryReturnStatement
 			return;
 		}

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/module/subscriber/SubscriptionMatchingSubscriber.java

@@ -1,6 +1,8 @@
 package ca.uhn.fhir.jpa.subscription.module.subscriber;
 
 import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.jpa.model.interceptor.api.IInterceptorRegistry;
+import ca.uhn.fhir.jpa.model.interceptor.api.Pointcut;
 import ca.uhn.fhir.jpa.subscription.module.ResourceModifiedMessage;
 import ca.uhn.fhir.jpa.subscription.module.cache.ActiveSubscription;
 import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionRegistry;
@@ -31,9 +33,9 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
  * 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.
@@ -52,6 +54,8 @@ public class SubscriptionMatchingSubscriber implements MessageHandler {
 	private FhirContext myFhirContext;
 	@Autowired
 	private SubscriptionRegistry mySubscriptionRegistry;
+	@Autowired
+	private IInterceptorRegistry myInterceptorRegistry;
 
 	@Override
 	public void handleMessage(Message<?> theMessage) throws MessagingException {
@@ -64,9 +68,18 @@ public class SubscriptionMatchingSubscriber implements MessageHandler {
 
 		ResourceModifiedMessage msg = ((ResourceModifiedJsonMessage) theMessage).getPayload();
 		matchActiveSubscriptionsAndDeliver(msg);
+
 	}
 
 	public void matchActiveSubscriptionsAndDeliver(ResourceModifiedMessage theMsg) {
+		try {
+			doMatchActiveSubscriptionsAndDeliver(theMsg);
+		} finally {
+			myInterceptorRegistry.callHooks(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED, theMsg);
+		}
+	}
+
+	private void doMatchActiveSubscriptionsAndDeliver(ResourceModifiedMessage theMsg) {
 		switch (theMsg.getOperationType()) {
 			case CREATE:
 			case UPDATE:

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/BaseSubscriptionTest.java

@@ -1,10 +1,12 @@
 package ca.uhn.fhir.jpa.subscription.module;
 
+import ca.uhn.fhir.jpa.model.interceptor.api.IInterceptorRegistry;
 import ca.uhn.fhir.jpa.searchparam.registry.ISearchParamRegistry;
 import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionLoader;
 import ca.uhn.fhir.jpa.subscription.module.config.MockFhirClientSearchParamProvider;
 import ca.uhn.fhir.jpa.subscription.module.config.MockFhirClientSubscriptionProvider;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
+import org.junit.After;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -24,6 +26,15 @@ public abstract class BaseSubscriptionTest {
 	@Autowired
 	SubscriptionLoader mySubscriptionLoader;
 
+	@Autowired
+	protected
+	IInterceptorRegistry myInterceptorRegistry;
+
+	@After
+	public void afterClearAnonymousLambdas() {
+		myInterceptorRegistry.clearAnonymousHookForUnitTest();
+	}
+
 	public void initSearchParamRegistry(IBundleProvider theBundleProvider) {
 		myMockFhirClientSearchParamProvider.setBundleProvider(theBundleProvider);
 		mySearchParamRegistry.forceRefresh();

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/config/TestSubscriptionDstu3Config.java

@@ -1,5 +1,7 @@
 package ca.uhn.fhir.jpa.subscription.module.config;
 
+import ca.uhn.fhir.jpa.model.interceptor.api.IInterceptorRegistry;
+import ca.uhn.fhir.jpa.model.interceptor.executor.InterceptorRegistry;
 import ca.uhn.fhir.jpa.searchparam.registry.ISearchParamProvider;
 import ca.uhn.fhir.jpa.subscription.module.cache.ISubscriptionProvider;
 import org.springframework.context.annotation.Bean;
@@ -18,5 +20,13 @@ public class TestSubscriptionDstu3Config extends SubscriptionDstu3Config {
 
 	@Bean
 	@Primary
-	public ISubscriptionProvider subsriptionProvider() { return new MockFhirClientSubscriptionProvider();}
+	public ISubscriptionProvider subsriptionProvider() {
+		return new MockFhirClientSubscriptionProvider();
+	}
+
+	@Bean
+	public IInterceptorRegistry interceptorRegistry() {
+		return new InterceptorRegistry();
+	}
+
 }

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/standalone/SubscriptionLoaderFhirClientTest.java

@@ -1,5 +1,6 @@
 package ca.uhn.fhir.jpa.subscription.module.standalone;
 
+import ca.uhn.fhir.jpa.model.interceptor.api.Pointcut;
 import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
 import ca.uhn.fhir.rest.server.SimpleBundleProvider;
@@ -8,12 +9,19 @@ import org.junit.Test;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
+import static org.hamcrest.Matchers.lessThan;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
 public class SubscriptionLoaderFhirClientTest extends BaseBlockingQueueSubscribableChannelDstu3Test {
 	@Test
-	public void testSubscriptionLoaderFhirClient() throws Exception {
+	public void testSubscriptionLoaderFhirClient() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		myInterceptorRegistry.registerAnonymousHookForUnitTest(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED, t-> latch.countDown());
+
 		String payload = "application/fhir+json";
 
 		String criteria1 = "Observation?code=SNOMED-CT|" + myCode + "&_format=xml";
@@ -27,6 +35,7 @@ public class SubscriptionLoaderFhirClientTest extends BaseBlockingQueueSubscriba
 		initSubscriptionLoader(bundle);
 
 		sendObservation(myCode, "SNOMED-CT");
+		latch.await(10, TimeUnit.SECONDS);
 
 		waitForSize(0, ourCreatedObservations);
 		waitForSize(1, ourUpdatedObservations);
@@ -34,7 +43,10 @@ public class SubscriptionLoaderFhirClientTest extends BaseBlockingQueueSubscriba
 	}
 
 	@Test
-	public void testSubscriptionLoaderFhirClientSubscriptionNotActive() throws Exception {
+	public void testSubscriptionLoaderFhirClientSubscriptionNotActive() throws InterruptedException {
+		CountDownLatch latch = new CountDownLatch(1);
+		myInterceptorRegistry.registerAnonymousHookForUnitTest(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED, t-> latch.countDown());
+
 		String payload = "application/fhir+json";
 
 		String criteria1 = "Observation?code=SNOMED-CT|" + myCode + "&_format=xml";
@@ -48,6 +60,7 @@ public class SubscriptionLoaderFhirClientTest extends BaseBlockingQueueSubscriba
 		initSubscriptionLoader(bundle);
 
 		sendObservation(myCode, "SNOMED-CT");
+		latch.await(10, TimeUnit.SECONDS);
 
 		waitForSize(0, ourCreatedObservations);
 		waitForSize(0, ourUpdatedObservations);