Merge branch 'master' into ja-subscription-interceptors

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

@@ -65,7 +65,11 @@ public class VersionUtil {
 
 			ourBuildNumber = p.getProperty("hapifhir.buildnumber");
 			ourBuildTime = p.getProperty("hapifhir.timestamp");
+
+			if (System.getProperty("suppress_hapi_fhir_version_log") == null) {
 				ourLog.info("HAPI FHIR version {} - Rev {}", ourVersion, StringUtils.right(ourBuildNumber, 10));
+			}
+
 		} catch (Exception e) {
 			ourLog.warn("Unable to determine HAPI version information", e);
 		}

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/config/TestR4Config.java

@@ -3,7 +3,6 @@ package ca.uhn.fhir.jpa.config;
 import ca.uhn.fhir.rest.server.interceptor.RequestValidatingInterceptor;
 import ca.uhn.fhir.validation.ResultSeverityEnum;
 import net.ttddyy.dsproxy.listener.SingleQueryCountHolder;
-import net.ttddyy.dsproxy.listener.logging.SLF4JLogLevel;
 import net.ttddyy.dsproxy.support.ProxyDataSourceBuilder;
 import org.apache.commons.dbcp2.BasicDataSource;
 import org.springframework.context.annotation.Bean;
@@ -97,7 +96,7 @@ public class TestR4Config extends BaseJavaConfigR4 {
 
 		DataSource dataSource = ProxyDataSourceBuilder
 			.create(retVal)
-			.logQueryBySlf4j(SLF4JLogLevel.INFO, "SQL")
+//			.logQueryBySlf4j(SLF4JLogLevel.INFO, "SQL")
 //			.logSlowQueryBySlf4j(10, TimeUnit.SECONDS)
 //			.countQuery(new ThreadQueryCountHolder())
 			.beforeQuery(new BlockLargeNumbersOfParamsListener())

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

@@ -89,17 +89,23 @@ public abstract class BaseSubscriptionsR4Test extends BaseResourceProviderR4Test
 		ourHeaders.clear();
 
 		// Delete all Subscriptions
+		if (ourClient != null) {
 			Bundle allSubscriptions = ourClient.search().forResource(Subscription.class).returnBundle(Bundle.class).execute();
 			for (IBaseResource next : BundleUtil.toListOfResources(myFhirCtx, allSubscriptions)) {
 				ourClient.delete().resource(next).execute();
 			}
 			waitForActivatedSubscriptionCount(0);
+		}
 
 		LinkedBlockingQueueSubscribableChannel processingChannel = mySubscriptionMatcherInterceptor.getProcessingChannelForUnitTest();
+		if (processingChannel != null) {
 			processingChannel.clearInterceptorsForUnitTest();
+		}
 		myCountingInterceptor = new CountingInterceptor();
+		if (processingChannel != null) {
 			processingChannel.addInterceptorForUnitTest(myCountingInterceptor);
 		}
+	}
 
 
 	protected Subscription createSubscription(String theCriteria, String thePayload) {

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

@@ -0,0 +1,39 @@
+package ca.uhn.fhir.jpa.subscription;
+
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Receives subscription notification without payloads.
+ */
+public class NotificationServlet extends HttpServlet {
+	private static final long serialVersionUID = 5957950857980374719L;
+
+	private final AtomicLong receivedNotificationCount = new AtomicLong();
+
+	private final List<String> receivedAuthorizationHeaders = Collections.synchronizedList(new ArrayList<>());
+
+	@Override
+	protected void doPost(HttpServletRequest req, HttpServletResponse resp) {
+		receivedNotificationCount.incrementAndGet();
+		receivedAuthorizationHeaders.add(req.getHeader("Authorization"));
+	}
+
+	public long getReceivedNotificationCount() {
+		return receivedNotificationCount.get();
+	}
+
+	public List<String> getReceivedAuthorizationHeaders() {
+		return receivedAuthorizationHeaders;
+	}
+
+	public void reset() {
+		receivedNotificationCount.set(0);
+		receivedAuthorizationHeaders.clear();
+	}
+}

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

@@ -4,6 +4,7 @@ package ca.uhn.fhir.jpa.subscription.resthook;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.jpa.dao.DaoConfig;
 import ca.uhn.fhir.jpa.provider.dstu3.BaseResourceProviderDstu3Test;
+import ca.uhn.fhir.jpa.subscription.NotificationServlet;
 import ca.uhn.fhir.jpa.subscription.SubscriptionTestUtil;
 import ca.uhn.fhir.rest.annotation.Create;
 import ca.uhn.fhir.rest.annotation.ResourceParam;
@@ -49,6 +50,8 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 
 	@Autowired
 	private SubscriptionTestUtil mySubscriptionTestUtil;
+	private static NotificationServlet ourNotificationServlet;
+	private static String ourNotificationListenerServer;
 
 	@After
 	public void afterUnregisterRestHookListener() {
@@ -79,9 +82,15 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		ourCreatedObservations.clear();
 		ourUpdatedObservations.clear();
 		ourContentTypes.clear();
+		ourNotificationServlet.reset();
 	}
 
-	private Subscription createSubscription(String theCriteria, String thePayload, String theEndpoint) throws InterruptedException {
+	private Subscription createSubscription(String criteria, String payload, String endpoint) throws InterruptedException {
+		return createSubscription(criteria, payload, endpoint, null);
+	}
+
+	private Subscription createSubscription(String theCriteria, String thePayload, String theEndpoint,
+														 List<StringType> headers) throws InterruptedException {
 		Subscription subscription = new Subscription();
 		subscription.setReason("Monitor new neonatal function (note, age will be determined by the monitor)");
 		subscription.setStatus(Subscription.SubscriptionStatus.REQUESTED);
@@ -91,6 +100,9 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		channel.setType(Subscription.SubscriptionChannelType.RESTHOOK);
 		channel.setPayload(thePayload);
 		channel.setEndpoint(theEndpoint);
+		if (headers != null) {
+			channel.setHeader(headers);
+		}
 		subscription.setChannel(channel);
 
 		MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
@@ -120,6 +132,55 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		return observation;
 	}
 
+	@Test
+	public void testRestHookSubscription() throws Exception {
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		createSubscription(criteria1, null, ourNotificationListenerServer,
+			Collections.singletonList(new StringType("Authorization: abc-def")));
+		createSubscription(criteria2, null, ourNotificationListenerServer);
+
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification with authorization header
+		waitForSize(1, ourNotificationServlet.getReceivedAuthorizationHeaders());
+		Assert.assertEquals(1, ourNotificationServlet.getReceivedNotificationCount());
+		Assert.assertEquals("abc-def", ourNotificationServlet.getReceivedAuthorizationHeaders().get(0));
+		ourNotificationServlet.reset();
+
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification with authorization header
+		waitForSize(1, ourNotificationServlet.getReceivedAuthorizationHeaders());
+		Assert.assertEquals(1, ourNotificationServlet.getReceivedNotificationCount());
+		Assert.assertEquals("abc-def", ourNotificationServlet.getReceivedAuthorizationHeaders().get(0));
+		ourNotificationServlet.reset();
+
+		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
+
+		/// Should see 1 subscription notification with authorization header
+		waitForSize(1, ourNotificationServlet.getReceivedAuthorizationHeaders());
+		Assert.assertEquals(1, ourNotificationServlet.getReceivedNotificationCount());
+		Assert.assertEquals("abc-def", ourNotificationServlet.getReceivedAuthorizationHeaders().get(0));
+		ourNotificationServlet.reset();
+
+		Observation observation3 = ourClient.read(Observation.class, observationTemp3.getId());
+		CodeableConcept codeableConcept = new CodeableConcept();
+		observation3.setCode(codeableConcept);
+		Coding coding = codeableConcept.addCoding();
+		coding.setCode(code + "111");
+		coding.setSystem("SNOMED-CT");
+		ourClient.update().resource(observation3).withId(observation3.getIdElement()).execute();
+
+		// Should see 2 subscription notifications with and without authorization header
+		waitForSize(1, ourNotificationServlet.getReceivedAuthorizationHeaders());
+		Assert.assertEquals(1, ourNotificationServlet.getReceivedNotificationCount());
+		Assert.assertNull(ourNotificationServlet.getReceivedAuthorizationHeaders().get(0));
+		ourNotificationServlet.reset();
+	}
+
 	@Test
 	public void testRestHookSubscriptionApplicationFhirJson() throws Exception {
 		String payload = "application/fhir+json";
@@ -358,11 +419,13 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		ourListenerPort = PortUtil.findFreePort();
 		ourListenerRestServer = new RestfulServer(FhirContext.forDstu3());
 		ourListenerServerBase = "http://localhost:" + ourListenerPort + "/fhir/context";
+		ourNotificationListenerServer = "http://localhost:" + ourListenerPort + "/fhir/subscription";
 
 		ObservationListener obsListener = new ObservationListener();
 		ourListenerRestServer.setResourceProviders(obsListener);
 
 		ourListenerServer = new Server(ourListenerPort);
+		ourNotificationServlet = new NotificationServlet();
 
 		ServletContextHandler proxyHandler = new ServletContextHandler();
 		proxyHandler.setContextPath("/");
@@ -370,6 +433,9 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		ServletHolder servletHolder = new ServletHolder();
 		servletHolder.setServlet(ourListenerRestServer);
 		proxyHandler.addServlet(servletHolder, "/fhir/context/*");
+		servletHolder = new ServletHolder();
+		servletHolder.setServlet(ourNotificationServlet);
+		proxyHandler.addServlet(servletHolder, "/fhir/subscription");
 
 		ourListenerServer.setHandler(proxyHandler);
 		ourListenerServer.start();

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

@@ -0,0 +1,165 @@
+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.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;
+import org.hl7.fhir.r4.model.IdType;
+import org.hl7.fhir.r4.model.Observation;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+
+import static org.hamcrest.Matchers.hasItem;
+import static org.junit.Assert.*;
+
+/**
+ * Test the rest-hook subscriptions
+ */
+@ContextConfiguration(classes = {RestHookWithInterceptorR4Test.MyCtxConfig.class})
+public class RestHookWithInterceptorR4Test extends BaseSubscriptionsR4Test {
+
+	private static final Logger ourLog = LoggerFactory.getLogger(RestHookWithInterceptorR4Test.class);
+	private static boolean ourNextModifyResourceId;
+	private static boolean ourNextBeforeRestHookDeliveryReturn;
+	private static boolean ourHitBeforeRestHookDelivery;
+	private static boolean ourNextAfterRestHookDeliveryReturn;
+	private static boolean ourHitAfterRestHookDelivery;
+	private static boolean ourNextAddHeader;
+	private static FhirContext ourCtx = FhirContext.forR4();
+
+	@Autowired
+	StoppableSubscriptionDeliveringRestHookSubscriber myStoppableSubscriptionDeliveringRestHookSubscriber;
+
+	@After
+	public void cleanupStoppableSubscriptionDeliveringRestHookSubscriber() {
+		myStoppableSubscriptionDeliveringRestHookSubscriber.setCountDownLatch(null);
+		myStoppableSubscriptionDeliveringRestHookSubscriber.unPause();
+	}
+
+	@Override
+	@Before
+	public void before() throws Exception {
+		super.before();
+		ourNextModifyResourceId = false;
+		ourNextAddHeader = false;
+		ourNextBeforeRestHookDeliveryReturn = true;
+		ourNextAfterRestHookDeliveryReturn = true;
+		ourHitBeforeRestHookDelivery = false;
+		ourHitAfterRestHookDelivery = false;
+	}
+
+	@Test
+	public void testBeforeRestHookDelivery_ModifyResourceId() throws Exception {
+		ourNextModifyResourceId = true;
+
+		createSubscription("Observation?status=final", "application/fhir+json");
+		waitForActivatedSubscriptionCount(1);
+
+		sendObservation();
+
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+		assertEquals("Observation/A", ourUpdatedObservations.get(0).getId());
+		assertTrue(ourHitBeforeRestHookDelivery);
+		assertTrue(ourHitAfterRestHookDelivery);
+	}
+
+	@Test
+	public void testBeforeRestHookDelivery_AddHeader() throws Exception {
+		ourNextAddHeader = true;
+
+		createSubscription("Observation?status=final", "application/fhir+json");
+		waitForActivatedSubscriptionCount(1);
+
+		sendObservation();
+
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+		assertTrue(ourHitBeforeRestHookDelivery);
+		assertTrue(ourHitAfterRestHookDelivery);
+		assertThat(ourHeaders, hasItem("X-Foo: Bar"));
+	}
+
+
+	@Test
+	public void testBeforeRestHookDelivery_AbortDelivery() throws Exception {
+		ourNextBeforeRestHookDeliveryReturn = false;
+
+		createSubscription("Observation?status=final", "application/fhir+json");
+		waitForActivatedSubscriptionCount(1);
+
+		sendObservation();
+
+		Thread.sleep(1000);
+		assertEquals(0, ourUpdatedObservations.size());
+	}
+
+	protected Observation sendObservation() {
+		Observation observation = new Observation();
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+		MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
+		observation.setId(methodOutcome.getId());
+		return observation;
+	}
+
+	@Configuration
+	public static class MyCtxConfig {
+
+		@Bean
+		public MyInterceptor interceptor() {
+			return new MyInterceptor();
+		}
+
+	}
+
+	/**
+	 * Interceptor class
+	 */
+	@SubscriptionInterceptor
+	public static class MyInterceptor {
+
+		/**
+		 * Constructor
+		 */
+		public MyInterceptor() {
+			ourLog.info("Creating interceptor");
+		}
+
+		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		public boolean beforeRestHookDelivery(ResourceDeliveryMessage theDeliveryMessage, CanonicalSubscription theSubscription) {
+			if (ourNextModifyResourceId) {
+				theDeliveryMessage.getPayload(ourCtx).setId(new IdType("Observation/A"));
+			}
+			if (ourNextAddHeader) {
+				theSubscription.addHeader("X-Foo: Bar");
+			}
+
+			ourHitBeforeRestHookDelivery = true;
+			return ourNextBeforeRestHookDeliveryReturn;
+		}
+
+		@SubscriptionHook(Pointcut.AFTER_REST_HOOK_DELIVERY)
+		public boolean afterRestHookDelivery(ResourceDeliveryMessage theDeliveryMessage, CanonicalSubscription theSubscription) {
+			ourHitAfterRestHookDelivery = true;
+			return ourNextAfterRestHookDeliveryReturn;
+		}
+
+	}
+
+
+}

hapi-fhir-jpaserver-model/pom.xml

@@ -72,12 +72,48 @@
 			<groupId>org.hibernate</groupId>
 			<artifactId>hibernate-search-orm</artifactId>
 		</dependency>
+
+		<!-- Spring -->
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-beans</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+			<exclusions>
+				<exclusion>
+					<artifactId>xml-apis</artifactId>
+					<groupId>xml-apis</groupId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-annotations</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.jscience</groupId>
+			<artifactId>jscience</artifactId>
+		</dependency>
+
+		<!-- Java -->
+		<dependency>
+			<groupId>javax.annotation</groupId>
+			<artifactId>javax.annotation-api</artifactId>
+		</dependency>
+
 		<!-- test dependencies -->
 		<dependency>
 			<groupId>ch.qos.logback</groupId>
 			<artifactId>logback-classic</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 
 	</dependencies>
 	<build>

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

@@ -0,0 +1,28 @@
+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/api/SubscriptionHook.java

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

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

@@ -0,0 +1,14 @@
+package ca.uhn.fhir.jpa.model.subscription.interceptor.api;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * This annotation declares a bean as a subscription interceptor
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface SubscriptionInterceptor {
+}

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

@@ -0,0 +1,48 @@
+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

@@ -0,0 +1,17 @@
+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-model/src/main/java/ca/uhn/fhir/jpa/model/subscription/interceptor/executor/SubscriptionInterceptorRegistry.java

@@ -0,0 +1,159 @@
+package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
+
+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 com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ListMultimap;
+import org.apache.commons.lang3.Validate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.annotation.Order;
+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.Collection;
+import java.util.IdentityHashMap;
+import java.util.List;
+
+@Component
+public class SubscriptionInterceptorRegistry implements ISubscriptionInterceptorRegistry, ApplicationContextAware {
+	private static final Logger ourLog = LoggerFactory.getLogger(SubscriptionInterceptorRegistry.class);
+	private ApplicationContext myAppCtx;
+	private List<Object> myGlobalInterceptors = new ArrayList<>();
+	private ListMultimap<Pointcut, Invoker> myInvokers = ArrayListMultimap.create();
+
+	List<Object> getGlobalInterceptors() {
+		return myGlobalInterceptors;
+	}
+
+	@PostConstruct
+	public void start() {
+
+		// Grab the global interceptors
+		String[] globalInterceptorNames = myAppCtx.getBeanNamesForAnnotation(SubscriptionInterceptor.class);
+		for (String nextName : globalInterceptorNames) {
+			Object nextGlobalInterceptor = myAppCtx.getBean(nextName);
+			myGlobalInterceptors.add(nextGlobalInterceptor);
+		}
+
+		// Sort them
+		sortByOrderAnnotation(myGlobalInterceptors);
+
+		// Pull out the hook methods
+		for (Object nextInterceptor : myGlobalInterceptors) {
+			for (Method nextMethod : nextInterceptor.getClass().getDeclaredMethods()) {
+				SubscriptionHook hook = AnnotationUtils.findAnnotation(nextMethod, SubscriptionHook.class);
+				if (hook != null) {
+					Invoker invoker = new Invoker(nextInterceptor, nextMethod);
+					for (Pointcut nextPointcut : hook.value()) {
+						myInvokers.put(nextPointcut, invoker);
+					}
+				}
+			}
+		}
+
+	}
+
+	private void sortByOrderAnnotation(List<Object> theObjects) {
+		IdentityHashMap<Object, Integer> interceptorToOrder = new IdentityHashMap<>();
+		for (Object next : theObjects) {
+			Order orderAnnotation = next.getClass().getAnnotation(Order.class);
+			int order = orderAnnotation != null ? orderAnnotation.value() : 0;
+			interceptorToOrder.put(next, order);
+		}
+
+		theObjects.sort((a, b) -> {
+			Integer orderA = interceptorToOrder.get(a);
+			Integer orderB = interceptorToOrder.get(b);
+			return orderA - orderB;
+		});
+	}
+
+	@Override
+	public void setApplicationContext(@Nonnull ApplicationContext theApplicationContext) throws BeansException {
+		myAppCtx = theApplicationContext;
+	}
+
+	@Override
+	public boolean callHooks(Pointcut thePointcut, HookParams theParams) {
+
+		/*
+		 * Call each hook in order
+		 */
+		List<Invoker> invokers = myInvokers.get(thePointcut);
+		for (Invoker nextInvoker : invokers) {
+			boolean shouldContinue = nextInvoker.invoke(theParams);
+			if (!shouldContinue) {
+				return false;
+			}
+		}
+
+		return true;
+	}
+
+	@Override
+	public boolean callHooks(Pointcut thePointcut, Object... theParams) {
+		return callHooks(thePointcut, new HookParams(theParams));
+	}
+
+	private class Invoker {
+
+		private final Object myInterceptor;
+		private final boolean myReturnsBoolean;
+		private final Method myMethod;
+		private final Class<?>[] myParameterTypes;
+
+		/**
+		 * Constructor
+		 */
+		private Invoker(@Nonnull Object theInterceptor, @Nonnull Method theHookMethod) {
+			myInterceptor = theInterceptor;
+			myParameterTypes = theHookMethod.getParameterTypes();
+			myMethod = theHookMethod;
+
+			Class<?> returnType = theHookMethod.getReturnType();
+			if (returnType.equals(boolean.class) || returnType.equals(Boolean.class)) {
+				myReturnsBoolean = true;
+			} else {
+				Validate.isTrue(Void.class.equals(returnType), "Method does not return boolean or void: %s", theHookMethod);
+				myReturnsBoolean = false;
+			}
+		}
+
+		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);
+				args[i] = nextParamValue;
+			}
+
+			// Invoke the method
+			try {
+				Object returnValue = myMethod.invoke(myInterceptor, args);
+				if (myReturnsBoolean) {
+					return (boolean) returnValue;
+				} else {
+					return true;
+				}
+			} catch (Exception e) {
+				ourLog.error("Failure executing interceptor method[{}]: {}", myMethod, e.toString(), e);
+				return true;
+			}
+
+		}
+
+	}
+
+	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/test/java/ca/uhn/fhir/jpa/model/subscription/interceptor/executor/SubscriptionInterceptorRegistryTest.java

@@ -0,0 +1,157 @@
+package ca.uhn.fhir.jpa.model.subscription.interceptor.executor;
+
+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 org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.annotation.Order;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.Matchers.contains;
+import static org.junit.Assert.*;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = {SubscriptionInterceptorRegistryTest.MyCtxConfig.class})
+public class SubscriptionInterceptorRegistryTest {
+
+	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;
+
+	@Autowired
+	private SubscriptionInterceptorRegistry mySubscriptionInterceptorRegistry;
+
+	@Test
+	public void testGlobalInterceptorsAreFound() {
+		List<Object> globalInterceptors = mySubscriptionInterceptorRegistry.getGlobalInterceptors();
+		assertEquals(2, globalInterceptors.size());
+		assertTrue(globalInterceptors.get(0).getClass().toString(), globalInterceptors.get(0) instanceof MyInterceptorOne);
+		assertTrue(globalInterceptors.get(1).getClass().toString(), globalInterceptors.get(1) instanceof MyInterceptorTwo);
+	}
+
+	@Test
+	public void testInvokeGlobalInterceptorMethods() {
+		ResourceDeliveryMessage msg = new ResourceDeliveryMessage();
+		CanonicalSubscription subs = new CanonicalSubscription();
+		HookParams params = new HookParams(msg, subs);
+		boolean outcome = mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+		assertTrue(outcome);
+
+		assertThat(ourInvocations, contains("MyInterceptorOne.beforeRestHookDelivery", "MyInterceptorTwo.beforeRestHookDelivery"));
+		assertSame(msg, ourLastResourceDeliveryMessage);
+		assertSame(subs, ourLastCanonicalSubscription);
+	}
+
+	@Test
+	public void testInvokeGlobalInterceptorMethods_MethodAbortsProcessing() {
+		ourNext_beforeRestHookDelivery_Return1 = false;
+
+		ResourceDeliveryMessage msg = new ResourceDeliveryMessage();
+		CanonicalSubscription subs = new CanonicalSubscription();
+		HookParams params = new HookParams(msg, subs);
+		boolean outcome = mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+		assertFalse(outcome);
+
+		assertThat(ourInvocations, contains("MyInterceptorOne.beforeRestHookDelivery"));
+	}
+
+	@Test
+	public void testCallHooksInvokedWithWrongParameters() {
+		Integer msg = 123;
+		CanonicalSubscription subs = new CanonicalSubscription();
+		HookParams params = new HookParams(msg, subs);
+		try {
+			mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, params);
+			fail();
+		} catch (AssertionError e) {
+			// good
+		}
+	}
+
+
+	@Before
+	public void before() {
+		ourNext_beforeRestHookDelivery_Return1 = true;
+		ourNext_beforeRestHookDelivery_Return2 = true;
+		ourLastCanonicalSubscription = null;
+		ourLastResourceDeliveryMessage = null;
+		ourInvocations.clear();
+	}
+
+	@Configuration
+	@ComponentScan(basePackages = "ca.uhn.fhir.jpa.model")
+	public static class MyCtxConfig {
+
+		@Bean
+		public SubscriptionInterceptorRegistry subscriptionInterceptorRegistry() {
+			return new SubscriptionInterceptorRegistry();
+		}
+
+		/**
+		 * Note: Orders are deliberately reversed to make sure we get the orders right
+		 * using the @Order annotation
+		 */
+		@Bean
+		public MyInterceptorTwo interceptor1() {
+			return new MyInterceptorTwo();
+		}
+
+		/**
+		 * 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
+	@Order(100)
+	public static class MyInterceptorOne {
+
+		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		public boolean beforeRestHookDelivery(CanonicalSubscription theCanonicalSubscription) {
+			ourLastCanonicalSubscription = theCanonicalSubscription;
+			ourInvocations.add("MyInterceptorOne.beforeRestHookDelivery");
+			return ourNext_beforeRestHookDelivery_Return1;
+		}
+
+	}
+
+	@SubscriptionInterceptor
+	@Order(200)
+	public static class MyInterceptorTwo {
+		@SubscriptionHook(Pointcut.BEFORE_REST_HOOK_DELIVERY)
+		public boolean beforeRestHookDelivery(ResourceDeliveryMessage theResourceDeliveryMessage) {
+			ourLastResourceDeliveryMessage = theResourceDeliveryMessage;
+			ourInvocations.add("MyInterceptorTwo.beforeRestHookDelivery");
+			return ourNext_beforeRestHookDelivery_Return2;
+		}
+	}
+
+	/**
+	 * Just a make-believe version of this class for the unit test
+	 */
+	private static class CanonicalSubscription {
+	}
+
+	/**
+	 * Just a make-believe version of this class for the unit test
+	 */
+	private static class ResourceDeliveryMessage {
+	}
+}

hapi-fhir-jpaserver-searchparam/pom.xml

@@ -87,36 +87,6 @@
 			<artifactId>javax.annotation-api</artifactId>
 		</dependency>
 
-		<!-- Spring -->
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-beans</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>org.springframework</groupId>
-			<artifactId>spring-context</artifactId>
-			<exclusions>
-				<exclusion>
-					<artifactId>xml-apis</artifactId>
-					<groupId>xml-apis</groupId>
-				</exclusion>
-			</exclusions>
-		</dependency>
-		<dependency>
-			<groupId>com.fasterxml.jackson.core</groupId>
-			<artifactId>jackson-annotations</artifactId>
-		</dependency>
-		<dependency>
-			<groupId>org.jscience</groupId>
-			<artifactId>jscience</artifactId>
-		</dependency>
-
-		<!-- Java -->
-		<dependency>
-			<groupId>javax.annotation</groupId>
-			<artifactId>javax.annotation-api</artifactId>
-		</dependency>
-
 		<!-- Testing -->
 
 		<dependency>

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

@@ -28,7 +28,6 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.hl7.fhir.instance.model.api.IIdType;
 import org.hl7.fhir.instance.model.api.IPrimitiveType;
-import org.hl7.fhir.r4.model.EventDefinition;
 import org.hl7.fhir.r4.model.Subscription;
 
 import javax.annotation.Nonnull;
@@ -39,7 +38,7 @@ import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
-public class CanonicalSubscription implements Serializable {
+public class CanonicalSubscription implements Serializable, Cloneable {
 
 	private static final long serialVersionUID = 1L;
 
@@ -73,32 +72,6 @@ public class CanonicalSubscription implements Serializable {
 		super();
 	}
 
-	/**
-	 * Copy Constructor
-	 */
-	public CanonicalSubscription(CanonicalSubscription theSubscription) {
-		myChannelExtensions = new HashMap<>(theSubscription.getChannelExtensions());
-		myChannelType = theSubscription.getChannelType();
-		myCriteriaString = theSubscription.getCriteriaString();
-		if (theSubscription.getEmailDetails() != null) {
-			myEmailDetails = new EmailDetails(theSubscription.getEmailDetails());
-		}
-		if (theSubscription.getRestHookDetails() != null) {
-			myRestHookDetails = new RestHookDetails(theSubscription.getRestHookDetails());
-		}
-		myHeaders = new ArrayList<>(theSubscription.getHeaders());
-		myEndpointUrl = theSubscription.getEndpointUrl();
-		myIdElement = theSubscription.getIdElementString();
-		myPayloadString = theSubscription.getPayloadString();
-		myStatus = theSubscription.getStatus();
-		// FIXME: is this constructor needed
-//		my
-	}
-
-	private Map<String, String> getChannelExtensions() {
-		return myChannelExtensions != null ? myChannelExtensions : Collections.emptyMap();
-	}
-
 	/**
 	 * For now we're using the R4 TriggerDefinition, but this
 	 * may change in the future when things stabilize
@@ -263,9 +236,28 @@ public class CanonicalSubscription implements Serializable {
 		}
 	}
 
+	/**
+	 * Adds a header
+	 *
+	 * @param theHeader The header, e.g. "Authorization: Bearer AAAAA"
+	 */
+	public void addHeader(String theHeader) {
+		if (isNotBlank(theHeader)) {
+			initHeaders();
+			myHeaders.add(theHeader);
+		}
+	}
+
+	private void initHeaders() {
+		if (myHeaders == null) {
+			myHeaders = new ArrayList<>();
+		}
+	}
+
 	@JsonInclude(JsonInclude.Include.NON_NULL)
 	@JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
 	public static class EmailDetails {
+
 		@JsonProperty("from")
 		private String myFrom;
 		@JsonProperty("subjectTemplate")
@@ -278,14 +270,6 @@ public class CanonicalSubscription implements Serializable {
 			super();
 		}
 
-		/**
-		 * Copy Constructor
-		 */
-		public EmailDetails(EmailDetails theEmailDetails) {
-			myFrom = theEmailDetails.getFrom();
-			mySubjectTemplate = theEmailDetails.getSubjectTemplate();
-		}
-
 		public String getFrom() {
 			return myFrom;
 		}
@@ -306,6 +290,7 @@ public class CanonicalSubscription implements Serializable {
 	@JsonInclude(JsonInclude.Include.NON_NULL)
 	@JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
 	public static class RestHookDetails {
+
 		@JsonProperty("stripVersionId")
 		private boolean myStripVersionId;
 		@JsonProperty("deliverLatestVersion")
@@ -318,14 +303,6 @@ public class CanonicalSubscription implements Serializable {
 			super();
 		}
 
-		/**
-		 * Copy constructor
-		 */
-		public RestHookDetails(RestHookDetails theRestHookDetails) {
-			myDeliverLatestVersion = theRestHookDetails.isDeliverLatestVersion();
-			myStripVersionId = theRestHookDetails.isStripVersionId();
-		}
-
 		public boolean isDeliverLatestVersion() {
 			return myDeliverLatestVersion;
 		}
@@ -370,9 +347,13 @@ public class CanonicalSubscription implements Serializable {
 	@JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
 	public static class CanonicalEventDefinition {
 
-		public CanonicalEventDefinition(EventDefinition theDef) {
+		/**
+		 * Constructor
+		 */
+		public CanonicalEventDefinition() {
 			// nothing yet
 		}
-	}
+
+	}
 
 }

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

@@ -33,12 +33,11 @@ import org.hl7.fhir.instance.model.api.IIdType;
 
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
+@SuppressWarnings("WeakerAccess")
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonAutoDetect(creatorVisibility = JsonAutoDetect.Visibility.NONE, fieldVisibility = JsonAutoDetect.Visibility.NONE, getterVisibility = JsonAutoDetect.Visibility.NONE, isGetterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
 public class ResourceDeliveryMessage implements IResourceMessage {
 
-	private static final long serialVersionUID = 1L;
-
 	@JsonIgnore
 	private transient CanonicalSubscription mySubscription;
 	@JsonProperty("subscription")
@@ -52,10 +51,6 @@ public class ResourceDeliveryMessage implements IResourceMessage {
 	@JsonProperty("operationType")
 	private ResourceModifiedMessage.OperationTypeEnum myOperationType;
 
-	public ResourceModifiedMessage.OperationTypeEnum getOperationType() {
-		return myOperationType;
-	}
-
 	/**
 	 * Constructor
 	 */
@@ -63,6 +58,10 @@ public class ResourceDeliveryMessage implements IResourceMessage {
 		super();
 	}
 
+	public ResourceModifiedMessage.OperationTypeEnum getOperationType() {
+		return myOperationType;
+	}
+
 	public void setOperationType(ResourceModifiedMessage.OperationTypeEnum theOperationType) {
 		myOperationType = theOperationType;
 	}
@@ -104,14 +103,15 @@ public class ResourceDeliveryMessage implements IResourceMessage {
 		myPayloadId = thePayload.getIdElement().toUnqualified().getValue();
 	}
 
+	@Override
+	public String getPayloadId() {
+		return myPayloadId;
+	}
+
 	public void setPayloadId(IIdType thePayloadId) {
 		myPayloadId = null;
 		if (thePayloadId != null) {
 			myPayloadId = thePayloadId.getValue();
 		}
 	}
-	@Override
-	public String getPayloadId() {
-		return myPayloadId;
-	}
 }

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

@@ -21,6 +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.subscription.interceptor.executor.ISubscriptionInterceptorRegistry;
 import ca.uhn.fhir.rest.api.EncodingEnum;
 import ca.uhn.fhir.rest.api.RequestTypeEnum;
 import ca.uhn.fhir.rest.client.api.*;
@@ -28,6 +30,7 @@ import ca.uhn.fhir.rest.client.interceptor.SimpleRequestHeaderInterceptor;
 import ca.uhn.fhir.rest.gclient.IClientExecutable;
 import ca.uhn.fhir.rest.server.exceptions.ResourceGoneException;
 import ca.uhn.fhir.rest.server.exceptions.ResourceNotFoundException;
+import org.apache.commons.lang3.StringUtils;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.instance.model.api.IIdType;
 import org.slf4j.Logger;
@@ -38,20 +41,18 @@ import org.springframework.messaging.MessagingException;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
 @Component
 @Scope("prototype")
 public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDeliverySubscriber {
-	private Logger ourLog = LoggerFactory.getLogger(SubscriptionDeliveringRestHookSubscriber.class);
-
 	@Autowired
 	IResourceRetriever myResourceRetriever;
+	private Logger ourLog = LoggerFactory.getLogger(SubscriptionDeliveringRestHookSubscriber.class);
+	@Autowired
+	private ISubscriptionInterceptorRegistry mySubscriptionInterceptorRegistry;
 
 	protected void deliverPayload(ResourceDeliveryMessage theMsg, CanonicalSubscription theSubscription, EncodingEnum thePayloadType, IGenericClient theClient) {
 		IBaseResource payloadResource = getAndMassagePayload(theMsg, theSubscription);
@@ -131,6 +132,11 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 	public void handleMessage(ResourceDeliveryMessage theMessage) throws MessagingException {
 		CanonicalSubscription subscription = theMessage.getSubscription();
 
+		// Interceptor call: BEFORE_REST_HOOK_DELIVERY
+		if (!mySubscriptionInterceptorRegistry.callHooks(Pointcut.BEFORE_REST_HOOK_DELIVERY, theMessage, subscription)) {
+			return;
+		}
+
 		// Grab the endpoint from the subscription
 		String endpointUrl = subscription.getEndpointUrl();
 
@@ -161,6 +167,13 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 		}
 
 		deliverPayload(theMessage, subscription, payloadType, client);
+
+		// Interceptor call: AFTER_REST_HOOK_DELIVERY
+		if (!mySubscriptionInterceptorRegistry.callHooks(Pointcut.AFTER_REST_HOOK_DELIVERY, theMessage, subscription)) {
+			//noinspection UnnecessaryReturnStatement
+			return;
+		}
+
 	}
 
 	/**
@@ -169,11 +182,26 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
 	protected void sendNotification(ResourceDeliveryMessage theMsg) {
 		Map<String, List<String>> params = new HashMap<>();
 		List<Header> headers = new ArrayList<>();
+		if (theMsg.getSubscription().getHeaders() != null) {
+			theMsg.getSubscription().getHeaders().stream().filter(Objects::nonNull).forEach(h -> {
+				final int sep = h.indexOf(':');
+				if (sep > 0) {
+					final String name = h.substring(0, sep);
+					final String value = h.substring(sep + 1);
+					if (StringUtils.isNotBlank(name)) {
+						headers.add(new Header(name.trim(), value.trim()));
+					}
+				}
+			});
+		}
+
 		StringBuilder url = new StringBuilder(theMsg.getSubscription().getEndpointUrl());
 		IHttpClient client = myFhirContext.getRestfulClientFactory().getHttpClient(url, params, "", RequestTypeEnum.POST, headers);
 		IHttpRequest request = client.createParamRequest(myFhirContext, params, null);
 		try {
 			IHttpResponse response = request.execute();
+			// close connection in order to return a possible cached connection to the connection pool
+			response.close();
 		} catch (IOException e) {
 			ourLog.error("Error trying to reach " + theMsg.getSubscription().getEndpointUrl());
 			e.printStackTrace();

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

@@ -7,6 +7,7 @@ import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionRegistry;
 import ca.uhn.fhir.jpa.subscription.module.matcher.ISubscriptionMatcher;
 import ca.uhn.fhir.jpa.subscription.module.matcher.SubscriptionMatchResult;
 import org.apache.commons.lang3.StringUtils;
+import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.instance.model.api.IIdType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -121,11 +122,15 @@ public class SubscriptionMatchingSubscriber implements MessageHandler {
 
 			ourLog.info("Subscription {} was matched by resource {} using matcher {}", nextActiveSubscription.getSubscription().getIdElement(myFhirContext).getValue(), id.toUnqualifiedVersionless().getValue(), matchResult.matcherShortName());
 
+			IBaseResource payload = theMsg.getNewPayload(myFhirContext);
+
 			ResourceDeliveryMessage deliveryMsg = new ResourceDeliveryMessage();
-			deliveryMsg.setPayload(myFhirContext, theMsg.getNewPayload(myFhirContext));
+			deliveryMsg.setPayload(myFhirContext, payload);
 			deliveryMsg.setSubscription(nextActiveSubscription.getSubscription());
 			deliveryMsg.setOperationType(theMsg.getOperationType());
+			if (payload == null) {
 				deliveryMsg.setPayloadId(theMsg.getId(myFhirContext));
+			}
 
 			ResourceDeliveryJsonMessage wrappedMsg = new ResourceDeliveryJsonMessage(deliveryMsg);
 			MessageChannel deliveryChannel = nextActiveSubscription.getSubscribableChannel();

hapi-fhir-jpaserver-subscription/src/test/resources/logback-test.xml

@@ -0,0 +1,14 @@
+<configuration>
+
+	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} [%file:%line] - %msg%n
+			</pattern>
+		</encoder>
+	</appender>
+
+	<root level="info">
+		<appender-ref ref="STDOUT" />
+	</root>
+
+</configuration>

src/changes/changes.xml

@@ -303,6 +303,12 @@
 				AuthorizationInterceptor can now allow a user to perform a search that is scoped to a particular
 				resource (e.g. Patient?_id=123) if the user has read access for that specific instance.
 			</action>
+			<action type="fix" issue="1084">
+				In JPA Server REST Hook Subscriptions, any Headers defined in the
+				Subscription resource are now applied to the outgoing HTTP
+				request.
+				Thanks to Volker Schmidt for the pull request!
+			</action>
 			<action type="add">
 				HAPI FHIR will now log the Git revision when it first starts up (on the ame line as the version number
 				that it already logs).