Merge branch 'master' of github.com:jamesagnew/hapi-fhir

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/DaoRegistry.java

@@ -138,7 +138,7 @@ public class DaoRegistry implements ApplicationContextAware, IDaoRegistry {
 
 	@Override
 	public boolean isResourceTypeSupported(String theResourceType) {
-		return mySupportedResourceTypes.contains(theResourceType);
+		return mySupportedResourceTypes == null || mySupportedResourceTypes.contains(theResourceType);
 	}
 
 	private void init() {

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

@@ -0,0 +1,240 @@
+package ca.uhn.fhir.jpa.subscription;
+
+import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.jpa.dao.DaoConfig;
+import ca.uhn.fhir.jpa.provider.r5.BaseResourceProviderR5Test;
+import ca.uhn.fhir.jpa.subscription.module.LinkedBlockingQueueSubscribableChannel;
+import ca.uhn.fhir.rest.annotation.Create;
+import ca.uhn.fhir.rest.annotation.ResourceParam;
+import ca.uhn.fhir.rest.annotation.Update;
+import ca.uhn.fhir.rest.api.Constants;
+import ca.uhn.fhir.rest.api.MethodOutcome;
+import ca.uhn.fhir.rest.server.IResourceProvider;
+import ca.uhn.fhir.rest.server.RestfulServer;
+import ca.uhn.fhir.test.utilities.JettyUtil;
+import ca.uhn.fhir.util.BundleUtil;
+import com.google.common.collect.Lists;
+import net.ttddyy.dsproxy.QueryCount;
+import net.ttddyy.dsproxy.listener.SingleQueryCountHolder;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.servlet.ServletHolder;
+import org.hl7.fhir.instance.model.api.IBaseResource;
+import org.hl7.fhir.instance.model.api.IIdType;
+import org.hl7.fhir.r5.model.*;
+import org.junit.*;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.annotation.PostConstruct;
+import javax.servlet.http.HttpServletRequest;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+
+@Ignore
+public abstract class BaseSubscriptionsR5Test extends BaseResourceProviderR5Test {
+	private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(BaseSubscriptionsR5Test.class);
+
+	private static Server ourListenerServer;
+    protected static int ourListenerPort;
+	protected static List<String> ourContentTypes = Collections.synchronizedList(new ArrayList<>());
+	protected static List<String> ourHeaders = Collections.synchronizedList(new ArrayList<>());
+	private static SingleQueryCountHolder ourCountHolder;
+
+	@Autowired
+	private SingleQueryCountHolder myCountHolder;
+	@Autowired
+	protected SubscriptionTestUtil mySubscriptionTestUtil;
+	@Autowired
+	protected SubscriptionMatcherInterceptor mySubscriptionMatcherInterceptor;
+
+	protected CountingInterceptor myCountingInterceptor;
+
+	protected static List<Observation> ourCreatedObservations = Collections.synchronizedList(Lists.newArrayList());
+	protected static List<Observation> ourUpdatedObservations = Collections.synchronizedList(Lists.newArrayList());
+	private static String ourListenerServerBase;
+
+	protected List<IIdType> mySubscriptionIds = Collections.synchronizedList(new ArrayList<>());
+
+
+	@After
+	public void afterUnregisterRestHookListener() {
+		for (IIdType next : mySubscriptionIds) {
+			IIdType nextId = next.toUnqualifiedVersionless();
+			ourLog.info("Deleting: {}", nextId);
+			ourClient.delete().resourceById(nextId).execute();
+		}
+		mySubscriptionIds.clear();
+
+		myDaoConfig.setAllowMultipleDelete(true);
+		ourLog.info("Deleting all subscriptions");
+		ourClient.delete().resourceConditionalByUrl("Subscription?status=active").execute();
+		ourClient.delete().resourceConditionalByUrl("Observation?code:missing=false").execute();
+		ourLog.info("Done deleting all subscriptions");
+		myDaoConfig.setAllowMultipleDelete(new DaoConfig().isAllowMultipleDelete());
+
+		mySubscriptionTestUtil.unregisterSubscriptionInterceptor();
+	}
+
+	@Before
+	public void beforeRegisterRestHookListener() {
+		mySubscriptionTestUtil.registerRestHookInterceptor();
+	}
+
+	@Before
+	public void beforeReset() throws Exception {
+		ourCreatedObservations.clear();
+		ourUpdatedObservations.clear();
+		ourContentTypes.clear();
+		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) {
+		Subscription subscription = newSubscription(theCriteria, thePayload);
+
+		MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
+		subscription.setId(methodOutcome.getId().getIdPart());
+		mySubscriptionIds.add(methodOutcome.getId());
+
+		return subscription;
+	}
+
+	protected Subscription newSubscription(String theCriteria, String thePayload) {
+		Subscription subscription = new Subscription();
+		subscription.setReason("Monitor new neonatal function (note, age will be determined by the monitor)");
+		subscription.setStatus(Subscription.SubscriptionStatus.REQUESTED);
+		subscription.setCriteria(theCriteria);
+
+		Subscription.SubscriptionChannelComponent channel = subscription.getChannel();
+		channel.setType(Subscription.SubscriptionChannelType.RESTHOOK);
+		channel.setPayload(thePayload);
+		channel.setEndpoint(ourListenerServerBase);
+		return subscription;
+	}
+
+
+	protected void waitForQueueToDrain() throws InterruptedException {
+		mySubscriptionTestUtil.waitForQueueToDrain();
+	}
+
+	@PostConstruct
+	public void initializeOurCountHolder() {
+		ourCountHolder = myCountHolder;
+	}
+
+
+	protected Observation sendObservation(String code, String system) {
+		Observation observation = new Observation();
+		CodeableConcept codeableConcept = new CodeableConcept();
+		observation.setCode(codeableConcept);
+		observation.getIdentifierFirstRep().setSystem("foo").setValue("1");
+		Coding coding = codeableConcept.addCoding();
+		coding.setCode(code);
+		coding.setSystem(system);
+
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+
+		IIdType id = myObservationDao.create(observation).getId();
+		observation.setId(id);
+
+		return observation;
+	}
+
+
+
+	public static class ObservationListener implements IResourceProvider {
+
+		@Create
+		public MethodOutcome create(@ResourceParam Observation theObservation, HttpServletRequest theRequest) {
+			ourLog.info("Received Listener Create");
+			ourContentTypes.add(theRequest.getHeader(Constants.HEADER_CONTENT_TYPE).replaceAll(";.*", ""));
+			ourCreatedObservations.add(theObservation);
+			extractHeaders(theRequest);
+			return new MethodOutcome(new IdType("Observation/1"), true);
+		}
+
+		private void extractHeaders(HttpServletRequest theRequest) {
+			Enumeration<String> headerNamesEnum = theRequest.getHeaderNames();
+			while (headerNamesEnum.hasMoreElements()) {
+				String nextName = headerNamesEnum.nextElement();
+				Enumeration<String> valueEnum = theRequest.getHeaders(nextName);
+				while (valueEnum.hasMoreElements()) {
+					String nextValue = valueEnum.nextElement();
+					ourHeaders.add(nextName + ": " + nextValue);
+				}
+			}
+		}
+
+		@Override
+		public Class<? extends IBaseResource> getResourceType() {
+			return Observation.class;
+		}
+
+		@Update
+		public MethodOutcome update(@ResourceParam Observation theObservation, HttpServletRequest theRequest) {
+			ourLog.info("Received Listener Update");
+			ourUpdatedObservations.add(theObservation);
+			ourContentTypes.add(theRequest.getHeader(Constants.HEADER_CONTENT_TYPE).replaceAll(";.*", ""));
+			extractHeaders(theRequest);
+			return new MethodOutcome(new IdType("Observation/1"), false);
+		}
+
+	}
+
+	@AfterClass
+	public static void reportTotalSelects() {
+		ourLog.info("Total database select queries: {}", getQueryCount().getSelect());
+	}
+
+	private static QueryCount getQueryCount() {
+		return ourCountHolder.getQueryCountMap().get("");
+	}
+
+	@BeforeClass
+	public static void startListenerServer() throws Exception {
+		RestfulServer ourListenerRestServer = new RestfulServer(FhirContext.forR5());
+		
+		ObservationListener obsListener = new ObservationListener();
+		ourListenerRestServer.setResourceProviders(obsListener);
+
+		ourListenerServer = new Server(0);
+
+		ServletContextHandler proxyHandler = new ServletContextHandler();
+		proxyHandler.setContextPath("/");
+
+		ServletHolder servletHolder = new ServletHolder();
+		servletHolder.setServlet(ourListenerRestServer);
+		proxyHandler.addServlet(servletHolder, "/fhir/context/*");
+
+		ourListenerServer.setHandler(proxyHandler);
+		JettyUtil.startServer(ourListenerServer);
+        ourListenerPort = JettyUtil.getPortForStartedServer(ourListenerServer);
+        ourListenerServerBase = "http://localhost:" + ourListenerPort + "/fhir/context";
+	}
+
+	@AfterClass
+	public static void stopListenerServer() throws Exception {
+		JettyUtil.closeServer(ourListenerServer);
+	}
+
+}

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

@@ -0,0 +1,957 @@
+package ca.uhn.fhir.jpa.subscription.resthook;
+
+import ca.uhn.fhir.jpa.config.StoppableSubscriptionDeliveringRestHookSubscriber;
+import ca.uhn.fhir.jpa.model.util.JpaConstants;
+import ca.uhn.fhir.jpa.subscription.BaseSubscriptionsR5Test;
+import ca.uhn.fhir.rest.api.CacheControlDirective;
+import ca.uhn.fhir.rest.api.Constants;
+import ca.uhn.fhir.rest.api.MethodOutcome;
+import ca.uhn.fhir.rest.server.exceptions.UnprocessableEntityException;
+import org.hl7.fhir.instance.model.api.IBaseBundle;
+import org.hl7.fhir.r5.model.*;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.junit.Assert.*;
+
+/**
+ * Test the rest-hook subscriptions
+ */
+public class RestHookTestR5Test extends BaseSubscriptionsR5Test {
+	private static final Logger ourLog = LoggerFactory.getLogger(RestHookTestR5Test.class);
+
+	@Autowired
+	StoppableSubscriptionDeliveringRestHookSubscriber myStoppableSubscriptionDeliveringRestHookSubscriber;
+
+	@After
+	public void cleanupStoppableSubscriptionDeliveringRestHookSubscriber() {
+		ourLog.info("@After");
+		myStoppableSubscriptionDeliveringRestHookSubscriber.setCountDownLatch(null);
+		myStoppableSubscriptionDeliveringRestHookSubscriber.unPause();
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationFhirJson() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		createSubscription(criteria1, payload);
+		createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+	}
+
+	@Test
+	public void testUpdatesHaveCorrectMetadata() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?";
+
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		/*
+		 * Send version 1
+		 */
+
+		Observation obs = sendObservation(code, "SNOMED-CT");
+		obs = myObservationDao.read(obs.getIdElement().toUnqualifiedVersionless());
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		int idx = 0;
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(idx));
+		assertEquals("1", ourUpdatedObservations.get(idx).getIdElement().getVersionIdPart());
+		assertEquals("1", ourUpdatedObservations.get(idx).getMeta().getVersionId());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals("1", ourUpdatedObservations.get(idx).getIdentifierFirstRep().getValue());
+
+		/*
+		 * Send version 2
+		 */
+
+		obs.getIdentifierFirstRep().setSystem("foo").setValue("2");
+		myObservationDao.update(obs);
+		obs = myObservationDao.read(obs.getIdElement().toUnqualifiedVersionless());
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		idx++;
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(2, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(idx));
+		assertEquals("2", ourUpdatedObservations.get(idx).getIdElement().getVersionIdPart());
+		assertEquals("2", ourUpdatedObservations.get(idx).getMeta().getVersionId());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals("2", ourUpdatedObservations.get(idx).getIdentifierFirstRep().getValue());
+	}
+
+	@Test
+	public void testPlaceholderReferencesInTransactionAreResolvedCorrectly() throws Exception {
+
+		String payload = "application/fhir+json";
+		String code = "1000000050";
+		String criteria1 = "Observation?";
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		// Create a transaction that should match
+		Bundle bundle = new Bundle();
+		bundle.setType(Bundle.BundleType.TRANSACTION);
+
+		Patient patient = new Patient();
+		patient.setId(IdType.newRandomUuid());
+		patient.getIdentifierFirstRep().setSystem("foo").setValue("AAA");
+		bundle.addEntry().setResource(patient).getRequest().setMethod(Bundle.HTTPVerb.POST).setUrl("Patient");
+
+		Observation observation = new Observation();
+		observation.getIdentifierFirstRep().setSystem("foo").setValue("1");
+		observation.getCode().addCoding().setCode(code).setSystem("SNOMED-CT");
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+		observation.getSubject().setReference(patient.getId());
+		bundle.addEntry().setResource(observation).getRequest().setMethod(Bundle.HTTPVerb.POST).setUrl("Observation");
+
+		// Send the transaction
+		mySystemDao.transaction(null, bundle);
+
+		waitForSize(1, ourUpdatedObservations);
+
+		assertThat(ourUpdatedObservations.get(0).getSubject().getReference(), matchesPattern("Patient/[0-9]+"));
+	}
+
+	@Test
+	public void testUpdatesHaveCorrectMetadataUsingTransactions() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?";
+
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		/*
+		 * Send version 1
+		 */
+
+		Observation observation = new Observation();
+		observation.getIdentifierFirstRep().setSystem("foo").setValue("1");
+		observation.getCode().addCoding().setCode(code).setSystem("SNOMED-CT");
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+		Bundle bundle = new Bundle();
+		bundle.setType(Bundle.BundleType.TRANSACTION);
+		bundle.addEntry().setResource(observation).getRequest().setMethod(Bundle.HTTPVerb.POST).setUrl("Observation");
+		Bundle responseBundle = mySystemDao.transaction(null, bundle);
+
+		Observation obs = myObservationDao.read(new IdType(responseBundle.getEntry().get(0).getResponse().getLocation()));
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		int idx = 0;
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(idx));
+		assertEquals("1", ourUpdatedObservations.get(idx).getIdElement().getVersionIdPart());
+		assertEquals("1", ourUpdatedObservations.get(idx).getMeta().getVersionId());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals("1", ourUpdatedObservations.get(idx).getIdentifierFirstRep().getValue());
+
+		/*
+		 * Send version 2
+		 */
+
+		observation = new Observation();
+		observation.setId(obs.getId());
+		observation.getIdentifierFirstRep().setSystem("foo").setValue("2");
+		observation.getCode().addCoding().setCode(code).setSystem("SNOMED-CT");
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+		bundle = new Bundle();
+		bundle.setType(Bundle.BundleType.TRANSACTION);
+		bundle.addEntry().setResource(observation).getRequest().setMethod(Bundle.HTTPVerb.PUT).setUrl(obs.getIdElement().toUnqualifiedVersionless().getValue());
+		mySystemDao.transaction(null, bundle);
+		obs = myObservationDao.read(obs.getIdElement().toUnqualifiedVersionless());
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		idx++;
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(2, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(idx));
+		assertEquals("2", ourUpdatedObservations.get(idx).getIdElement().getVersionIdPart());
+		assertEquals("2", ourUpdatedObservations.get(idx).getMeta().getVersionId());
+		assertEquals(obs.getMeta().getLastUpdatedElement().getValueAsString(), ourUpdatedObservations.get(idx).getMeta().getLastUpdatedElement().getValueAsString());
+		assertEquals("2", ourUpdatedObservations.get(idx).getIdentifierFirstRep().getValue());
+	}
+
+	@Test
+	public void testRepeatedDeliveries() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?";
+
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		for (int i = 0; i < 100; i++) {
+			Observation observation = new Observation();
+			observation.getIdentifierFirstRep().setSystem("foo").setValue("ID" + i);
+			observation.getCode().addCoding().setCode(code).setSystem("SNOMED-CT");
+			observation.setStatus(Observation.ObservationStatus.FINAL);
+			myObservationDao.create(observation);
+		}
+
+		waitForSize(100, ourUpdatedObservations);
+	}
+
+	@Test
+	public void testActiveSubscriptionShouldntReActivate() throws Exception {
+		String criteria = "Observation?code=111111111&_format=xml";
+		String payload = "application/fhir+json";
+		createSubscription(criteria, payload);
+
+		waitForActivatedSubscriptionCount(1);
+		for (int i = 0; i < 5; i++) {
+			int changes = this.mySubscriptionLoader.doSyncSubscriptionsForUnitTest();
+			assertEquals(0, changes);
+		}
+	}
+
+	@Test
+	public void testRestHookSubscriptionNoopUpdateDoesntTriggerNewDelivery() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		createSubscription(criteria1, payload);
+		createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		Observation obs = sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+
+		// Send an update with no changes
+		obs.setId(obs.getIdElement().toUnqualifiedVersionless());
+		ourClient.update().resource(obs).execute();
+
+		// Should be no further deliveries
+		Thread.sleep(1000);
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+
+
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationJsonDisableVersionIdInDelivery() throws Exception {
+		String payload = "application/json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		waitForActivatedSubscriptionCount(0);
+		Subscription subscription1 = createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+
+		ourLog.info("** About to send observation");
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+
+		IdType idElement = ourUpdatedObservations.get(0).getIdElement();
+		assertEquals(observation1.getIdElement().getIdPart(), idElement.getIdPart());
+		// VersionId is present
+		assertEquals(observation1.getIdElement().getVersionIdPart(), idElement.getVersionIdPart());
+
+		subscription1
+			.getChannel()
+			.addExtension(JpaConstants.EXT_SUBSCRIPTION_RESTHOOK_STRIP_VERSION_IDS, new BooleanType("true"));
+		ourLog.info("** About to update subscription");
+
+		int modCount = myCountingInterceptor.getSentCount();
+		ourClient.update().resource(subscription1).execute();
+		waitForSize(modCount + 1, () -> myCountingInterceptor.getSentCount());
+
+		ourLog.info("** About to send observation");
+		Observation observation2 = sendObservation(code, "SNOMED-CT");
+
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(2, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(1));
+
+		idElement = ourUpdatedObservations.get(1).getIdElement();
+		assertEquals(observation2.getIdElement().getIdPart(), idElement.getIdPart());
+		// Now VersionId is stripped
+		assertEquals(null, idElement.getVersionIdPart());
+	}
+
+	@Test
+	public void testRestHookSubscriptionDoesntGetLatestVersionByDefault() throws Exception {
+		String payload = "application/json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		waitForActivatedSubscriptionCount(0);
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		myStoppableSubscriptionDeliveringRestHookSubscriber.pause();
+		final CountDownLatch countDownLatch = new CountDownLatch(1);
+		myStoppableSubscriptionDeliveringRestHookSubscriber.setCountDownLatch(countDownLatch);
+
+		ourLog.info("** About to send observation");
+		Observation observation = sendObservation(code, "SNOMED-CT");
+		assertEquals("1", observation.getIdElement().getVersionIdPart());
+		assertNull(observation.getNoteFirstRep().getText());
+
+		observation.getNoteFirstRep().setText("changed");
+		MethodOutcome methodOutcome = ourClient.update().resource(observation).execute();
+		assertEquals("2", methodOutcome.getId().getVersionIdPart());
+		assertEquals("changed", observation.getNoteFirstRep().getText());
+
+		// Wait for our two delivery channel threads to be paused
+		assertTrue(countDownLatch.await(5L, TimeUnit.SECONDS));
+		// Open the floodgates!
+		myStoppableSubscriptionDeliveringRestHookSubscriber.unPause();
+
+
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(2, ourUpdatedObservations);
+
+		Observation observation1 = ourUpdatedObservations.get(0);
+		Observation observation2 = ourUpdatedObservations.get(1);
+
+		assertEquals("1", observation1.getIdElement().getVersionIdPart());
+		assertNull(observation1.getNoteFirstRep().getText());
+		assertEquals("2", observation2.getIdElement().getVersionIdPart());
+		assertEquals("changed", observation2.getNoteFirstRep().getText());
+	}
+
+	@Test
+	public void testRestHookSubscriptionGetsLatestVersionWithFlag() throws Exception {
+		String payload = "application/json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		waitForActivatedSubscriptionCount(0);
+
+		Subscription subscription = newSubscription(criteria1, payload);
+		subscription
+			.getChannel()
+			.addExtension(JpaConstants.EXT_SUBSCRIPTION_RESTHOOK_DELIVER_LATEST_VERSION, new BooleanType("true"));
+		ourClient.create().resource(subscription).execute();
+
+		waitForActivatedSubscriptionCount(1);
+
+		myStoppableSubscriptionDeliveringRestHookSubscriber.pause();
+		final CountDownLatch countDownLatch = new CountDownLatch(1);
+		myStoppableSubscriptionDeliveringRestHookSubscriber.setCountDownLatch(countDownLatch);
+
+		ourLog.info("** About to send observation");
+		Observation observation = sendObservation(code, "SNOMED-CT");
+		assertEquals("1", observation.getIdElement().getVersionIdPart());
+		assertNull(observation.getNoteFirstRep().getText());
+
+		observation.getNoteFirstRep().setText("changed");
+		MethodOutcome methodOutcome = ourClient.update().resource(observation).execute();
+		assertEquals("2", methodOutcome.getId().getVersionIdPart());
+		assertEquals("changed", observation.getNoteFirstRep().getText());
+
+		// Wait for our two delivery channel threads to be paused
+		assertTrue(countDownLatch.await(5L, TimeUnit.SECONDS));
+		// Open the floodgates!
+		myStoppableSubscriptionDeliveringRestHookSubscriber.unPause();
+
+
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(2, ourUpdatedObservations);
+
+		Observation observation1 = ourUpdatedObservations.get(0);
+		Observation observation2 = ourUpdatedObservations.get(1);
+
+		assertEquals("2", observation1.getIdElement().getVersionIdPart());
+		assertEquals("changed", observation1.getNoteFirstRep().getText());
+		assertEquals("2", observation2.getIdElement().getVersionIdPart());
+		assertEquals("changed", observation2.getNoteFirstRep().getText());
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationJson() throws Exception {
+		String payload = "application/json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		Subscription subscription1 = createSubscription(criteria1, payload);
+		Subscription subscription2 = createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+
+		assertEquals("1", ourUpdatedObservations.get(0).getIdElement().getVersionIdPart());
+
+		Subscription subscriptionTemp = ourClient.read(Subscription.class, subscription2.getId());
+		Assert.assertNotNull(subscriptionTemp);
+
+		subscriptionTemp.setCriteria(criteria1);
+		ourClient.update().resource(subscriptionTemp).withId(subscriptionTemp.getIdElement()).execute();
+		waitForQueueToDrain();
+
+		Observation observation2 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see two subscription notifications
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(3, ourUpdatedObservations);
+
+		ourClient.delete().resourceById(new IdType("Subscription/" + subscription2.getId())).execute();
+		waitForQueueToDrain();
+
+		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see only one subscription notification
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		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 no subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		Observation observation3a = ourClient.read(Observation.class, observationTemp3.getId());
+
+		CodeableConcept codeableConcept1 = new CodeableConcept();
+		observation3a.setCode(codeableConcept1);
+		Coding coding1 = codeableConcept1.addCoding();
+		coding1.setCode(code);
+		coding1.setSystem("SNOMED-CT");
+		ourClient.update().resource(observation3a).withId(observation3a.getIdElement()).execute();
+
+		// Should see only one subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(5, ourUpdatedObservations);
+
+		assertFalse(subscription1.getId().equals(subscription2.getId()));
+		assertFalse(observation1.getId().isEmpty());
+		assertFalse(observation2.getId().isEmpty());
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationJsonDatabase() throws Exception {
+		// Same test as above, but now run it using database matching
+		myDaoConfig.setEnableInMemorySubscriptionMatching(false);
+		String payload = "application/json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		Subscription subscription1 = createSubscription(criteria1, payload);
+		Subscription subscription2 = createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+
+		assertEquals("1", ourUpdatedObservations.get(0).getIdElement().getVersionIdPart());
+
+		Subscription subscriptionTemp = ourClient.read(Subscription.class, subscription2.getId());
+		Assert.assertNotNull(subscriptionTemp);
+
+		subscriptionTemp.setCriteria(criteria1);
+		ourClient.update().resource(subscriptionTemp).withId(subscriptionTemp.getIdElement()).execute();
+		waitForQueueToDrain();
+
+		Observation observation2 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see two subscription notifications
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(3, ourUpdatedObservations);
+
+		ourClient.delete().resourceById(new IdType("Subscription/" + subscription2.getId())).execute();
+		waitForQueueToDrain();
+
+		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see only one subscription notification
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		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 no subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		Observation observation3a = ourClient.read(Observation.class, observationTemp3.getId());
+
+		CodeableConcept codeableConcept1 = new CodeableConcept();
+		observation3a.setCode(codeableConcept1);
+		Coding coding1 = codeableConcept1.addCoding();
+		coding1.setCode(code);
+		coding1.setSystem("SNOMED-CT");
+		ourClient.update().resource(observation3a).withId(observation3a.getIdElement()).execute();
+
+		// Should see only one subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(5, ourUpdatedObservations);
+
+		assertFalse(subscription1.getId().equals(subscription2.getId()));
+		assertFalse(observation1.getId().isEmpty());
+		assertFalse(observation2.getId().isEmpty());
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationXml() throws Exception {
+		String payload = "application/xml";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		Subscription subscription1 = createSubscription(criteria1, payload);
+		Subscription subscription2 = createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		ourLog.info("** About to send obervation");
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_XML_NEW, ourContentTypes.get(0));
+
+		Subscription subscriptionTemp = ourClient.read(Subscription.class, subscription2.getId());
+		Assert.assertNotNull(subscriptionTemp);
+		subscriptionTemp.setCriteria(criteria1);
+		ourClient.update().resource(subscriptionTemp).withId(subscriptionTemp.getIdElement()).execute();
+		waitForQueueToDrain();
+
+		Observation observation2 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see two subscription notifications
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(3, ourUpdatedObservations);
+
+		ourClient.delete().resourceById(new IdType("Subscription/" + subscription2.getId())).execute();
+
+		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
+
+		// Should see only one subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		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 no subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(4, ourUpdatedObservations);
+
+		Observation observation3a = ourClient.read(Observation.class, observationTemp3.getId());
+
+		CodeableConcept codeableConcept1 = new CodeableConcept();
+		observation3a.setCode(codeableConcept1);
+		Coding coding1 = codeableConcept1.addCoding();
+		coding1.setCode(code);
+		coding1.setSystem("SNOMED-CT");
+		ourClient.update().resource(observation3a).withId(observation3a.getIdElement()).execute();
+
+		// Should see only one subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(5, ourUpdatedObservations);
+
+		assertFalse(subscription1.getId().equals(subscription2.getId()));
+		assertFalse(observation1.getId().isEmpty());
+		assertFalse(observation2.getId().isEmpty());
+	}
+
+	@Test
+	public void testSubscriptionTriggerViaSubscription() throws Exception {
+		String payload = "application/xml";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		ourLog.info("** About to send obervation");
+
+		Observation observation = new Observation();
+		observation.addIdentifier().setSystem("foo").setValue("bar1");
+		observation.setId(IdType.newRandomUuid().getValue());
+		CodeableConcept codeableConcept = new CodeableConcept()
+			.addCoding(new Coding().setCode(code).setSystem("SNOMED-CT"));
+		observation.setCode(codeableConcept);
+		observation.setStatus(Observation.ObservationStatus.FINAL);
+
+		Patient patient = new Patient();
+		patient.addIdentifier().setSystem("foo").setValue("bar2");
+		patient.setId(IdType.newRandomUuid().getValue());
+		patient.setActive(true);
+		observation.getSubject().setReference(patient.getId());
+
+		Bundle requestBundle = new Bundle();
+		requestBundle.setType(Bundle.BundleType.TRANSACTION);
+		requestBundle.addEntry()
+			.setResource(observation)
+			.setFullUrl(observation.getId())
+			.getRequest()
+			.setUrl("Obervation?identifier=foo|bar1")
+			.setMethod(Bundle.HTTPVerb.PUT);
+		requestBundle.addEntry()
+			.setResource(patient)
+			.setFullUrl(patient.getId())
+			.getRequest()
+			.setUrl("Patient?identifier=foo|bar2")
+			.setMethod(Bundle.HTTPVerb.PUT);
+		ourClient.transaction().withBundle(requestBundle).execute();
+
+		// Should see 1 subscription notification
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_XML_NEW, ourContentTypes.get(0));
+
+		Observation obs = ourUpdatedObservations.get(0);
+		ourLog.info("Observation content: {}", myFhirCtx.newXmlParser().setPrettyPrint(true).encodeResourceToString(obs));
+	}
+
+	@Test
+	public void testUpdateSubscriptionToMatchLater() throws Exception {
+		String payload = "application/xml";
+
+		String code = "1000000050";
+		String criteriaBad = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		ourLog.info("** About to create non-matching subscription");
+
+		Subscription subscription2 = createSubscription(criteriaBad, payload);
+
+		ourLog.info("** About to send observation that wont match");
+
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		// Criteria didn't match, shouldn't see any updates
+		waitForQueueToDrain();
+		Thread.sleep(1000);
+		assertEquals(0, ourUpdatedObservations.size());
+
+		Subscription subscriptionTemp = ourClient.read().resource(Subscription.class).withId(subscription2.getId()).execute();
+		Assert.assertNotNull(subscriptionTemp);
+		String criteriaGood = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		subscriptionTemp.setCriteria(criteriaGood);
+		ourLog.info("** About to update subscription");
+		ourClient.update().resource(subscriptionTemp).withId(subscriptionTemp.getIdElement()).execute();
+		waitForQueueToDrain();
+
+		ourLog.info("** About to send Observation 2");
+		Observation observation2 = sendObservation(code, "SNOMED-CT");
+		waitForQueueToDrain();
+
+		// Should see a subscription notification this time
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+
+		ourClient.delete().resourceById(new IdType("Subscription/" + subscription2.getId())).execute();
+
+		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
+
+		// No more matches
+		Thread.sleep(1000);
+		assertEquals(1, ourUpdatedObservations.size());
+	}
+
+	@Test
+	public void testRestHookSubscriptionApplicationXmlJson() throws Exception {
+		String payload = "application/fhir+xml";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+		String criteria2 = "Observation?code=SNOMED-CT|" + code + "111&_format=xml";
+
+		Subscription subscription1 = createSubscription(criteria1, payload);
+		Subscription subscription2 = createSubscription(criteria2, payload);
+		waitForActivatedSubscriptionCount(2);
+
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_XML_NEW, ourContentTypes.get(0));
+	}
+
+	@Test
+	public void testRestHookSubscriptionInvalidCriteria() throws Exception {
+		String payload = "application/xml";
+
+		String criteria1 = "Observation?codeeeee=SNOMED-CT";
+
+		try {
+			createSubscription(criteria1, payload);
+			fail();
+		} catch (UnprocessableEntityException e) {
+			assertEquals("HTTP 422 Unprocessable Entity: Invalid subscription criteria submitted: Observation?codeeeee=SNOMED-CT Failed to parse match URL[Observation?codeeeee=SNOMED-CT] - Resource type Observation does not have a parameter with name: codeeeee", e.getMessage());
+		}
+	}
+
+	@Test
+	public void testSubscriptionWithHeaders() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		// Add some headers, and we'll also turn back to requested status for fun
+		Subscription subscription = createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		subscription.getChannel().addHeader("X-Foo: FOO");
+		subscription.getChannel().addHeader("X-Bar: BAR");
+		subscription.setStatus(Subscription.SubscriptionStatus.REQUESTED);
+		ourClient.update().resource(subscription).execute();
+		waitForQueueToDrain();
+
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+		assertEquals(Constants.CT_FHIR_JSON_NEW, ourContentTypes.get(0));
+		assertThat(ourHeaders, hasItem("X-Foo: FOO"));
+		assertThat(ourHeaders, hasItem("X-Bar: BAR"));
+	}
+
+	@Test
+	public void testDisableSubscription() throws Exception {
+		String payload = "application/fhir+json";
+
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		Subscription subscription = createSubscription(criteria1, payload);
+		waitForActivatedSubscriptionCount(1);
+
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+
+		// Disable
+		subscription.setStatus(Subscription.SubscriptionStatus.OFF);
+		ourClient.update().resource(subscription).execute();
+		waitForQueueToDrain();
+
+		// Send another object
+		sendObservation(code, "SNOMED-CT");
+
+		// Should see 1 subscription notification
+		waitForQueueToDrain();
+		waitForSize(0, ourCreatedObservations);
+		waitForSize(1, ourUpdatedObservations);
+
+	}
+
+	@Test(expected = UnprocessableEntityException.class)
+	public void testInvalidProvenanceParam() {
+		String payload = "application/fhir+json";
+		String criteriabad = "Provenance?activity=http://hl7.org/fhir/v3/DocumentCompletion%7CAU";
+		Subscription subscription = newSubscription(criteriabad, payload);
+		ourClient.create().resource(subscription).execute();
+	}
+
+	@Test(expected = UnprocessableEntityException.class)
+	public void testInvalidProcedureRequestParam() {
+		String payload = "application/fhir+json";
+		String criteriabad = "ProcedureRequest?intent=instance-order&category=Laboratory";
+		Subscription subscription = newSubscription(criteriabad, payload);
+		ourClient.create().resource(subscription).execute();
+	}
+
+	@Test(expected = UnprocessableEntityException.class)
+	public void testInvalidBodySiteParam() {
+		String payload = "application/fhir+json";
+		String criteriabad = "BodySite?accessType=Catheter";
+		Subscription subscription = newSubscription(criteriabad, payload);
+		ourClient.create().resource(subscription).execute();
+	}
+
+	@Test
+	public void testGoodSubscriptionPersists() {
+		assertEquals(0, subscriptionCount());
+		String payload = "application/fhir+json";
+		String criteriaGood = "Patient?gender=male";
+		Subscription subscription = newSubscription(criteriaGood, payload);
+		ourClient.create().resource(subscription).execute();
+		assertEquals(1, subscriptionCount());
+	}
+
+	private int subscriptionCount() {
+		IBaseBundle found = ourClient.search().forResource(Subscription.class).cacheControl(new CacheControlDirective().setNoCache(true)).execute();
+		return toUnqualifiedVersionlessIdValues(found).size();
+	}
+
+	@Test
+	public void testSubscriptionWithNoStatusIsRejected() {
+		Subscription subscription = newSubscription("Observation?", "application/json");
+		subscription.setStatus(null);
+
+		try {
+			ourClient.create().resource(subscription).execute();
+			fail();
+		} catch (UnprocessableEntityException e) {
+			assertThat(e.getMessage(), containsString("Can not process submitted Subscription - Subscription.status must be populated on this server"));
+		}
+	}
+
+
+	@Test
+	public void testBadSubscriptionDoesntPersist() {
+		assertEquals(0, subscriptionCount());
+		String payload = "application/fhir+json";
+		String criteriaBad = "BodySite?accessType=Catheter";
+		Subscription subscription = newSubscription(criteriaBad, payload);
+		try {
+			ourClient.create().resource(subscription).execute();
+		} catch (UnprocessableEntityException e) {
+			ourLog.info("Expected exception", e);
+		}
+		assertEquals(0, subscriptionCount());
+	}
+
+	@Test
+	public void testCustomSearchParam() throws Exception {
+		String criteria = "Observation?accessType=Catheter,PD%20Catheter";
+
+		SearchParameter sp = new SearchParameter();
+		sp.addBase("Observation");
+		sp.setCode("accessType");
+		sp.setType(Enumerations.SearchParamType.TOKEN);
+		sp.setExpression("Observation.extension('Observation#accessType')");
+		sp.setXpathUsage(SearchParameter.XPathUsageType.NORMAL);
+		sp.setStatus(Enumerations.PublicationStatus.ACTIVE);
+		mySearchParameterDao.create(sp);
+		mySearchParamRegistry.forceRefresh();
+		createSubscription(criteria, "application/json");
+		waitForActivatedSubscriptionCount(1);
+
+		{
+			Observation bodySite = new Observation();
+			bodySite.addExtension().setUrl("Observation#accessType").setValue(new Coding().setCode("Catheter"));
+			MethodOutcome methodOutcome = ourClient.create().resource(bodySite).execute();
+			assertEquals(true, methodOutcome.getCreated());
+			waitForQueueToDrain();
+			waitForSize(1, ourUpdatedObservations);
+		}
+		{
+			Observation observation = new Observation();
+			observation.addExtension().setUrl("Observation#accessType").setValue(new Coding().setCode("PD Catheter"));
+			MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
+			assertEquals(true, methodOutcome.getCreated());
+			waitForQueueToDrain();
+			waitForSize(2, ourUpdatedObservations);
+		}
+		{
+			Observation observation = new Observation();
+			MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
+			assertEquals(true, methodOutcome.getCreated());
+			waitForQueueToDrain();
+			waitForSize(2, ourUpdatedObservations);
+		}
+		{
+			Observation observation = new Observation();
+			observation.addExtension().setUrl("Observation#accessType").setValue(new Coding().setCode("XXX"));
+			MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
+			assertEquals(true, methodOutcome.getCreated());
+			waitForQueueToDrain();
+			waitForSize(2, ourUpdatedObservations);
+		}
+
+	}
+
+
+}

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

@@ -34,6 +34,8 @@ import org.hl7.fhir.instance.model.api.IBaseReference;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.instance.model.api.IPrimitiveType;
 import org.hl7.fhir.r4.model.Extension;
+import org.hl7.fhir.r4.model.codesystems.SubscriptionStatus;
+import org.hl7.fhir.r5.model.Subscription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -67,6 +69,8 @@ public class SubscriptionCanonicalizer<S extends IBaseResource> {
 				return canonicalizeDstu3(theSubscription);
 			case R4:
 				return canonicalizeR4(theSubscription);
+			case R5:
+				return canonicalizeR5(theSubscription);
 			case DSTU2_HL7ORG:
 			case DSTU2_1:
 			default:
@@ -169,6 +173,14 @@ public class SubscriptionCanonicalizer<S extends IBaseResource> {
 						.stream()
 						.collect(Collectors.groupingBy(t -> t.getUrl(), mapping(t -> t.getValueAsPrimitive().getValueAsString(), toList())));
 				}
+				case R5: {
+					org.hl7.fhir.r5.model.Subscription subscription = (org.hl7.fhir.r5.model.Subscription) theSubscription;
+					return subscription
+						.getChannel()
+						.getExtension()
+						.stream()
+						.collect(Collectors.groupingBy(t -> t.getUrl(), mapping(t -> t.getValueAsPrimitive().getValueAsString(), toList())));
+				}
 				case DSTU2_HL7ORG:
 				case DSTU2_1:
 				default: {
@@ -232,6 +244,56 @@ public class SubscriptionCanonicalizer<S extends IBaseResource> {
 		return retVal;
 	}
 
+	private CanonicalSubscription canonicalizeR5(IBaseResource theSubscription) {
+		org.hl7.fhir.r5.model.Subscription subscription = (org.hl7.fhir.r5.model.Subscription) theSubscription;
+
+		CanonicalSubscription retVal = new CanonicalSubscription();
+		retVal.setStatus(org.hl7.fhir.r4.model.Subscription.SubscriptionStatus.fromCode(subscription.getStatus().toCode()));
+		retVal.setChannelType(CanonicalSubscriptionChannelType.fromCode(subscription.getChannel().getType().toCode()));
+		retVal.setCriteriaString(subscription.getCriteria());
+		retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
+		retVal.setHeaders(subscription.getChannel().getHeader());
+		retVal.setChannelExtensions(extractExtension(subscription));
+		retVal.setIdElement(subscription.getIdElement());
+		retVal.setPayloadString(subscription.getChannel().getPayload());
+
+		if (retVal.getChannelType() == CanonicalSubscriptionChannelType.EMAIL) {
+			String from;
+			String subjectTemplate;
+			try {
+				from = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_EMAIL_FROM);
+				subjectTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_SUBJECT_TEMPLATE);
+			} catch (FHIRException theE) {
+				throw new ConfigurationException("Failed to extract subscription extension(s): " + theE.getMessage(), theE);
+			}
+			retVal.getEmailDetails().setFrom(from);
+			retVal.getEmailDetails().setSubjectTemplate(subjectTemplate);
+		}
+
+		if (retVal.getChannelType() == CanonicalSubscriptionChannelType.RESTHOOK) {
+			String stripVersionIds;
+			String deliverLatestVersion;
+			try {
+				stripVersionIds = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_RESTHOOK_STRIP_VERSION_IDS);
+				deliverLatestVersion = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_RESTHOOK_DELIVER_LATEST_VERSION);
+			} catch (FHIRException theE) {
+				throw new ConfigurationException("Failed to extract subscription extension(s): " + theE.getMessage(), theE);
+			}
+			retVal.getRestHookDetails().setStripVersionId(Boolean.parseBoolean(stripVersionIds));
+			retVal.getRestHookDetails().setDeliverLatestVersion(Boolean.parseBoolean(deliverLatestVersion));
+		}
+
+		List<org.hl7.fhir.r5.model.Extension> topicExts = subscription.getExtensionsByUrl("http://hl7.org/fhir/subscription/topics");
+		if (topicExts.size() > 0) {
+			IBaseReference ref = (IBaseReference) topicExts.get(0).getValueAsPrimitive();
+			if (!"EventDefinition".equals(ref.getReferenceElement().getResourceType())) {
+				throw new PreconditionFailedException("Topic reference must be an EventDefinition");
+			}
+		}
+
+		return retVal;
+	}
+
 	public String getCriteria(IBaseResource theSubscription) {
 		switch (myFhirContext.getVersion().getVersion()) {
 			case DSTU2:
@@ -240,6 +302,8 @@ public class SubscriptionCanonicalizer<S extends IBaseResource> {
 				return ((org.hl7.fhir.dstu3.model.Subscription) theSubscription).getCriteria();
 			case R4:
 				return ((org.hl7.fhir.r4.model.Subscription) theSubscription).getCriteria();
+			case R5:
+				return ((org.hl7.fhir.r5.model.Subscription) theSubscription).getCriteria();
 			case DSTU2_1:
 			case DSTU2_HL7ORG:
 			default:

src/changes/changes.xml

@@ -25,10 +25,29 @@
 					<li>commons-lang3 (Core): 3.8.1 -&gt; 3.9</li>
 					<li>commons-text (Core): 1.6 -&gt; 1.7</li>
 					<li>Guava (JPA): 27.1-jre -&gt; 28.0-jre</li>
-					<li>
 				</ul>
 				]]>
 			</action>
+			<action type="change">
+				<![CDATA[
+				<b>Breaking Change</b>:
+				The HL7.org DSTU2 structures (and <i>ONLY</i> the HL7.org DSTU2 structures) have been
+				moved to a new package. Where they were previously found in 
+				<code>org.hl7.fhir.instance.model</code>
+				they are now found in
+				<code>org.hl7.fhir.dstu2.model</code>. This was done in order to complete the harmonization
+				between the 
+				<a href="https://github.com/jamesagnew/hapi-fhir">HAPI FHIR</a>
+				GitHub repository and the 
+				<a href="https://github.com/hapifhir/org.hl7.fhir.core/">org.hl7.fhir.core</a>
+				GitHub repository. This is the kind of change we don't make lightly, as we do know that it
+				will be annoying for users of the existing library. It is a change however that will allow us
+				to apply validator fixes much more quickly, and will greatly reduce the amount of effort
+				required to keep up with R5 changes as they come out, so we're hoping it is worth it.
+				Note that no classes are removed, they have only been moved, so it should be fairly straightforward
+				to migrate existing code with an IDE.
+				]]>
+			</action>
 			<action type="change">
 				<![CDATA[
 				<b>Breaking Change</b>: