Refactor subscriptions to use a single codebase across versions

hapi-fhir-jpaserver-base/pom.xml

@@ -148,6 +148,15 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>org.thymeleaf</groupId>
+			<artifactId>thymeleaf</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.thymeleaf</groupId>
+			<artifactId>thymeleaf-spring4</artifactId>
+		</dependency>
+
 		<!-- For UCUM -->
 		<dependency>
 			<groupId>org.jscience</groupId>
@@ -344,6 +353,10 @@
 			<groupId>javax.mail</groupId>
 			<artifactId>javax.mail-api</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.sun.mail</groupId>
+			<artifactId>javax.mail</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>javax.validation</groupId>
 			<artifactId>validation-api</artifactId>
@@ -422,6 +435,23 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>com.icegreen</groupId>
+			<artifactId>greenmail</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>com.sun.mail</groupId>
+					<artifactId>javax.mail</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>com.icegreen</groupId>
+			<artifactId>greenmail-spring</artifactId>
+			<scope>test</scope>
+		</dependency>
+
 	</dependencies>
 
 

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

@@ -1606,7 +1606,7 @@ public class SearchBuilder implements ISearchBuilder {
 	public void loadResourcesByPid(Collection<Long> theIncludePids, List<IBaseResource> theResourceListToPopulate, Set<Long> theRevIncludedPids, boolean theForHistoryOperation,
 											 EntityManager entityManager, FhirContext context, IDao theDao) {
 		if (theIncludePids.isEmpty()) {
-			ourLog.info("The include pids are empty");
+			ourLog.debug("The include pids are empty");
 			// return;
 		}
 

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/BaseSubscriptionDeliverySubscriber.java

@@ -0,0 +1,42 @@
+package ca.uhn.fhir.jpa.subscription;
+
+import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
+import org.hl7.fhir.r4.model.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+
+public abstract class BaseSubscriptionDeliverySubscriber extends BaseSubscriptionSubscriber {
+	private static final Logger ourLog = LoggerFactory.getLogger(BaseSubscriptionDeliverySubscriber.class);
+
+	public BaseSubscriptionDeliverySubscriber(IFhirResourceDao<?> theSubscriptionDao, Subscription.SubscriptionChannelType theChannelType, BaseSubscriptionInterceptor theSubscriptionInterceptor) {
+		super(theSubscriptionDao, theChannelType, theSubscriptionInterceptor);
+	}
+
+	@Override
+	public void handleMessage(Message<?> theMessage) throws MessagingException {
+		if (!(theMessage.getPayload() instanceof ResourceDeliveryMessage)) {
+			return;
+		}
+		try {
+			ResourceDeliveryMessage msg = (ResourceDeliveryMessage) theMessage.getPayload();
+			if (!subscriptionTypeApplies(getContext(), msg.getSubscription().getBackingSubscription())) {
+				return;
+			}
+
+			CanonicalSubscription updatedSubscription = (CanonicalSubscription)getSubscriptionInterceptor().getIdToSubscription().get(msg.getSubscription().getIdElement().getIdPart());
+			if (updatedSubscription != null) {
+				msg.setSubscription(updatedSubscription);
+			}
+
+			handleMessage(msg);
+		} catch (Exception e) {
+			ourLog.error("Failure handling subscription payload", e);
+			throw new MessagingException(theMessage, "Failure handling subscription payload", e);
+		}
+	}
+
+	public abstract void handleMessage(ResourceDeliveryMessage theMessage) throws Exception;
+
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/BaseSubscriptionInterceptor.java

@@ -26,6 +26,7 @@ import ca.uhn.fhir.context.FhirVersionEnum;
 import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
 import ca.uhn.fhir.jpa.dao.SearchParameterMap;
 import ca.uhn.fhir.jpa.provider.ServletSubRequestDetails;
+import ca.uhn.fhir.jpa.util.JpaConstants;
 import ca.uhn.fhir.jpa.util.StopWatch;
 import ca.uhn.fhir.rest.api.RestOperationTypeEnum;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
@@ -51,6 +52,8 @@ import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.ExecutorSubscribableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.transaction.support.TransactionSynchronizationAdapter;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
@@ -85,7 +88,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
 	@Autowired(required = false)
 	@Qualifier("myEventDefinitionDaoR4")
 	private IFhirResourceDao<org.hl7.fhir.r4.model.EventDefinition> myEventDefinitionDaoR4;
-
 	/**
 	 * Constructor
 	 */
@@ -139,6 +141,23 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
 			retVal.setHeaders(subscription.getChannel().getHeader());
 			retVal.setIdElement(subscription.getIdElement());
 			retVal.setPayloadString(subscription.getChannel().getPayload());
+
+			if (retVal.getChannelType() == Subscription.SubscriptionChannelType.EMAIL) {
+				String from;
+				String subjectTemplate;
+				String bodyTemplate;
+				try {
+					from = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_EMAIL_FROM);
+					subjectTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_SUBJECT_TEMPLATE);
+					bodyTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_BODY_TEMPLATE);
+				} catch (FHIRException theE) {
+					throw new ConfigurationException("Failed to extract subscription extension(s): " + theE.getMessage(), theE);
+				}
+				retVal.getEmailDetails().setFrom(from);
+				retVal.getEmailDetails().setSubjectTemplate(subjectTemplate);
+				retVal.getEmailDetails().setBodyTemplate(bodyTemplate);
+			}
+
 		} catch (FHIRException theE) {
 			throw new InternalErrorException(theE);
 		}
@@ -158,6 +177,22 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
 		retVal.setIdElement(subscription.getIdElement());
 		retVal.setPayloadString(subscription.getChannel().getPayload());
 
+		if (retVal.getChannelType() == Subscription.SubscriptionChannelType.EMAIL) {
+			String from;
+			String subjectTemplate;
+			String bodyTemplate;
+			try {
+				from = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_EMAIL_FROM);
+				subjectTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_SUBJECT_TEMPLATE);
+				bodyTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_BODY_TEMPLATE);
+			} catch (FHIRException theE) {
+				throw new ConfigurationException("Failed to extract subscription extension(s): " + theE.getMessage(), theE);
+			}
+			retVal.getEmailDetails().setFrom(from);
+			retVal.getEmailDetails().setSubjectTemplate(subjectTemplate);
+			retVal.getEmailDetails().setBodyTemplate(bodyTemplate);
+		}
+
 		List<org.hl7.fhir.r4.model.Extension> topicExts = subscription.getExtensionsByUrl("http://hl7.org/fhir/subscription/topics");
 		if (topicExts.size() > 0) {
 			IBaseReference ref = (IBaseReference) topicExts.get(0).getValueAsPrimitive();
@@ -260,8 +295,78 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
 		}
 	}
 
+	@SuppressWarnings("unused")
+	@PreDestroy
+	public void preDestroy() {
+		getProcessingChannel().unsubscribe(mySubscriptionCheckingSubscriber);
+
+		unregisterDeliverySubscriber();
+	}
+
+	protected abstract void registerDeliverySubscriber();
+
+	public void registerSubscription(IIdType theId, S theSubscription) {
+		Validate.notNull(theId);
+		Validate.notBlank(theId.getIdPart());
+		Validate.notNull(theSubscription);
+
+		myIdToSubscription.put(theId.getIdPart(), canonicalize(theSubscription));
+	}
+
+	protected void registerSubscriptionCheckingSubscriber() {
+		if (mySubscriptionCheckingSubscriber == null) {
+			mySubscriptionCheckingSubscriber = new SubscriptionCheckingSubscriber(getSubscriptionDao(), getChannelType(), this);
+		}
+		getProcessingChannel().subscribe(mySubscriptionCheckingSubscriber);
+	}
+
+	@Override
+	public void resourceCreated(RequestDetails theRequest, IBaseResource theResource) {
+		ResourceModifiedMessage msg = new ResourceModifiedMessage();
+		msg.setId(theResource.getIdElement());
+		msg.setOperationType(RestOperationTypeEnum.CREATE);
+		msg.setNewPayload(theResource);
+		submitResourceModified(msg);
+	}
+
+	@Override
+	public void resourceDeleted(RequestDetails theRequest, IBaseResource theResource) {
+		ResourceModifiedMessage msg = new ResourceModifiedMessage();
+		msg.setId(theResource.getIdElement());
+		msg.setOperationType(RestOperationTypeEnum.DELETE);
+		submitResourceModified(msg);
+	}
+
+	@Override
+	public void resourceUpdated(RequestDetails theRequest, IBaseResource theOldResource, IBaseResource theNewResource) {
+		ResourceModifiedMessage msg = new ResourceModifiedMessage();
+		msg.setId(theNewResource.getIdElement());
+		msg.setOperationType(RestOperationTypeEnum.UPDATE);
+		msg.setNewPayload(theNewResource);
+		submitResourceModified(msg);
+	}
+
+	public void setFhirContext(FhirContext theCtx) {
+		myCtx = theCtx;
+	}
+
+	public void setResourceDaos(List<IFhirResourceDao<?>> theResourceDaos) {
+		myResourceDaos = theResourceDaos;
+	}
+
 	@PostConstruct
-	public void postConstruct() {
+	public void start() {
+		for (IFhirResourceDao<?> next : myResourceDaos) {
+			if (myCtx.getResourceDefinition(next.getResourceType()).getName().equals("Subscription")) {
+				mySubscriptionDao = next;
+			}
+		}
+		Validate.notNull(mySubscriptionDao);
+
+		if (myCtx.getVersion().getVersion() == FhirVersionEnum.R4) {
+			Validate.notNull(myEventDefinitionDaoR4);
+		}
+
 		if (getProcessingChannel() == null) {
 			myProcessingExecutorQueue = new LinkedBlockingQueue<>(1000);
 			RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
@@ -336,75 +441,25 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
 		initSubscriptions();
 	}
 
-	@SuppressWarnings("unused")
-	@PreDestroy
-	public void preDestroy() {
-		getProcessingChannel().unsubscribe(mySubscriptionCheckingSubscriber);
-
-		unregisterDeliverySubscriber();
-	}
-
-	protected abstract void registerDeliverySubscriber();
-
-	public void registerSubscription(IIdType theId, S theSubscription) {
-		Validate.notNull(theId);
-		Validate.notBlank(theId.getIdPart());
-		Validate.notNull(theSubscription);
-
-		myIdToSubscription.put(theId.getIdPart(), canonicalize(theSubscription));
-	}
-
-	protected void registerSubscriptionCheckingSubscriber() {
-		if (mySubscriptionCheckingSubscriber == null) {
-			mySubscriptionCheckingSubscriber = new SubscriptionCheckingSubscriber(getSubscriptionDao(), getChannelType(), this);
-		}
-		getProcessingChannel().subscribe(mySubscriptionCheckingSubscriber);
-	}
-
-	@Override
-	public void resourceCreated(RequestDetails theRequest, IBaseResource theResource) {
-		ResourceModifiedMessage msg = new ResourceModifiedMessage();
-		msg.setId(theResource.getIdElement());
-		msg.setOperationType(RestOperationTypeEnum.CREATE);
-		msg.setNewPayload(theResource);
-		submitResourceModified(msg);
-	}
-
-	@Override
-	public void resourceDeleted(RequestDetails theRequest, IBaseResource theResource) {
-		ResourceModifiedMessage msg = new ResourceModifiedMessage();
-		msg.setId(theResource.getIdElement());
-		msg.setOperationType(RestOperationTypeEnum.DELETE);
-		submitResourceModified(msg);
-	}
-
-	@Override
-	public void resourceUpdated(RequestDetails theRequest, IBaseResource theOldResource, IBaseResource theNewResource) {
-		ResourceModifiedMessage msg = new ResourceModifiedMessage();
-		msg.setId(theNewResource.getIdElement());
-		msg.setOperationType(RestOperationTypeEnum.UPDATE);
-		msg.setNewPayload(theNewResource);
-		submitResourceModified(msg);
-	}
-
-	@PostConstruct
-	public void start() {
-		for (IFhirResourceDao<?> next : myResourceDaos) {
-			if (myCtx.getResourceDefinition(next.getResourceType()).getName().equals("Subscription")) {
-				mySubscriptionDao = next;
-			}
-		}
-		Validate.notNull(mySubscriptionDao);
-
-		if (myCtx.getVersion().getVersion() == FhirVersionEnum.R4) {
-			Validate.notNull(myEventDefinitionDaoR4);
-		}
-	}
-
 	protected void submitResourceModified(final ResourceModifiedMessage theMsg) {
 		final GenericMessage<ResourceModifiedMessage> message = new GenericMessage<>(theMsg);
 		mySubscriptionActivatingSubscriber.handleMessage(message);
-		getProcessingChannel().send(message);
+		sendToProcessingChannel(message);
+	}
+
+	protected void sendToProcessingChannel(final GenericMessage<ResourceModifiedMessage> theMessage) {
+		ourLog.trace("Registering synchronization to send resource modified message to processing channel");
+
+		/*
+		 * We only actually submit this item work working after the
+		 */
+		TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
+			@Override
+			public void afterCommit() {
+				ourLog.trace("Sending resource modified message to processing channel");
+				getProcessingChannel().send(theMessage);
+			}
+		});
 	}
 
 	protected abstract void unregisterDeliverySubscriber();

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/BaseSubscriptionSubscriber.java

@@ -26,6 +26,7 @@ import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.instance.model.api.IPrimitiveType;
 import org.hl7.fhir.r4.model.Subscription;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
 
 public abstract class BaseSubscriptionSubscriber implements MessageHandler {
 
@@ -58,6 +59,7 @@ public abstract class BaseSubscriptionSubscriber implements MessageHandler {
 		return mySubscriptionInterceptor;
 	}
 
+
 	/**
 	 * Does this subscription type (e.g. rest hook, websocket, etc) apply to this interceptor?
 	 */

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

@@ -47,6 +47,7 @@ public class CanonicalSubscription implements Serializable {
 	private Subscription.SubscriptionStatus myStatus;
 	private IBaseResource myBackingSubscription;
 	private TriggerDefinition myTrigger;
+	private EmailDetails myEmailDetails;
 
 	/**
 	 * For now we're using the R4 TriggerDefinition, but this
@@ -93,6 +94,13 @@ public class CanonicalSubscription implements Serializable {
 		myCriteriaString = theCriteriaString;
 	}
 
+	public EmailDetails getEmailDetails() {
+		if (myEmailDetails == null) {
+			myEmailDetails = new EmailDetails();
+		}
+		return myEmailDetails;
+	}
+
 	public String getEndpointUrl() {
 		return myEndpointUrl;
 	}
@@ -159,4 +167,35 @@ public class CanonicalSubscription implements Serializable {
 			myHeaders.add(theHeaders);
 		}
 	}
+
+	public static class EmailDetails {
+		private String myFrom;
+		private String mySubjectTemplate;
+		private String myBodyTemplate;
+
+		public String getBodyTemplate() {
+			return myBodyTemplate;
+		}
+
+		public void setBodyTemplate(String theBodyTemplate) {
+			myBodyTemplate = theBodyTemplate;
+		}
+
+		public String getFrom() {
+			return myFrom;
+		}
+
+		public void setFrom(String theFrom) {
+			myFrom = theFrom;
+		}
+
+		public String getSubjectTemplate() {
+			return mySubjectTemplate;
+		}
+
+		public void setSubjectTemplate(String theSubjectTemplate) {
+			mySubjectTemplate = theSubjectTemplate;
+		}
+	}
+
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/SubscriptionCheckingSubscriber.java

@@ -50,7 +50,10 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 
 	@Override
 	public void handleMessage(Message<?> theMessage) throws MessagingException {
+		ourLog.trace("Handling resource modified message: {}", theMessage);
+
 		if (!(theMessage.getPayload() instanceof ResourceModifiedMessage)) {
+			ourLog.warn("Unexpected message payload type: {}", theMessage.getPayload());
 			return;
 		}
 
@@ -60,6 +63,7 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 			case UPDATE:
 				break;
 			default:
+				ourLog.trace("Not processing modified message for {}", msg.getOperationType());
 				// ignore anything else
 				return;
 		}
@@ -68,6 +72,9 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 		String resourceId = msg.getId().getIdPart();
 
 		List<CanonicalSubscription> subscriptions = getSubscriptionInterceptor().getSubscriptions();
+
+		ourLog.trace("Testing {} subscriptions for applicability");
+
 		for (CanonicalSubscription nextSubscription : subscriptions) {
 
 			String nextSubscriptionId = nextSubscription.getIdElement().toUnqualifiedVersionless().getValue();
@@ -78,8 +85,7 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 			}
 
 			// see if the criteria matches the created object
-			ourLog.debug("Checking subscription {} for {} with criteria {}", nextSubscriptionId, resourceType, nextCriteriaString);
+			ourLog.trace("Checking subscription {} for {} with criteria {}", nextSubscriptionId, resourceType, nextCriteriaString);
-
 			String criteriaResource = nextCriteriaString;
 			int index = criteriaResource.indexOf("?");
 			if (index != -1) {
@@ -87,7 +93,7 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 			}
 
 			if (resourceType != null && nextCriteriaString != null && !criteriaResource.equals(resourceType)) {
-				ourLog.debug("Skipping subscription search for {} because it does not match the criteria {}", resourceType, nextCriteriaString);
+				ourLog.trace("Skipping subscription search for {} because it does not match the criteria {}", resourceType, nextCriteriaString);
 				continue;
 			}
 
@@ -97,6 +103,9 @@ public class SubscriptionCheckingSubscriber extends BaseSubscriptionSubscriber {
 			criteria = massageCriteria(criteria);
 
 			IBundleProvider results = performSearch(criteria);
+
+			ourLog.info("Subscription check found {} results for query: {}", results.size(), criteria);
+
 			if (results.size() == 0) {
 				continue;
 			}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/email/EmailDetails.java

@@ -0,0 +1,43 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+import java.util.List;
+
+public class EmailDetails {
+	private String mySubjectTemplate;
+	private String myBodyTemplate;
+	private List<String> myTo;
+	private String myFrom;
+
+	public String getBodyTemplate() {
+		return myBodyTemplate;
+	}
+
+	public void setBodyTemplate(String theBodyTemplate) {
+		myBodyTemplate = theBodyTemplate;
+	}
+
+	public String getFrom() {
+		return myFrom;
+	}
+
+	public void setFrom(String theFrom) {
+		myFrom = theFrom;
+	}
+
+	public String getSubjectTemplate() {
+		return mySubjectTemplate;
+	}
+
+	public void setSubjectTemplate(String theSubjectTemplate) {
+		mySubjectTemplate = theSubjectTemplate;
+	}
+
+	public List<String> getTo() {
+		return myTo;
+	}
+
+	public void setTo(List<String> theTo) {
+		myTo = theTo;
+	}
+
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/email/EmailSender.java

@@ -0,0 +1,98 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+import ca.uhn.fhir.jpa.util.StopWatch;
+import ca.uhn.fhir.rest.api.Constants;
+import org.apache.commons.lang3.Validate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.mail.SimpleMailMessage;
+import org.springframework.mail.javamail.JavaMailSender;
+import org.springframework.mail.javamail.JavaMailSenderImpl;
+import org.thymeleaf.context.Context;
+import org.thymeleaf.spring4.SpringTemplateEngine;
+import org.thymeleaf.spring4.dialect.SpringStandardDialect;
+import org.thymeleaf.templatemode.TemplateMode;
+import org.thymeleaf.templateresolver.StringTemplateResolver;
+
+import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import static org.apache.commons.lang3.StringUtils.trim;
+
+public class EmailSender implements IEmailSender {
+
+	private static final Logger ourLog = LoggerFactory.getLogger(EmailSender.class);
+	private String mySmtpServerHost;
+	private int mySmtpServerPort = 25;
+	private JavaMailSenderImpl mySender;
+
+	@PostConstruct
+	public void start() {
+		Validate.notBlank(mySmtpServerHost, "No SMTP host defined");
+
+		mySender = new JavaMailSenderImpl();
+		mySender.setHost(mySmtpServerHost);
+		mySender.setPort(mySmtpServerPort);
+		mySender.setDefaultEncoding(Constants.CHARSET_UTF8.name());
+	}
+
+	@Override
+	public void send(EmailDetails theDetails) {
+		ourLog.info("Sending email to recipients: {}", theDetails.getTo());
+		StopWatch sw = new StopWatch();
+
+		StringTemplateResolver templateResolver = new StringTemplateResolver();
+		templateResolver.setTemplateMode(TemplateMode.TEXT);
+
+		SpringStandardDialect dialect = new SpringStandardDialect();
+		dialect.setEnableSpringELCompiler(true);
+
+		SpringTemplateEngine engine = new SpringTemplateEngine();
+		engine.setDialect(dialect);
+		engine.setEnableSpringELCompiler(true);
+		engine.setTemplateResolver(templateResolver);
+
+		Context context = new Context();
+
+		String body = engine.process(theDetails.getBodyTemplate(), context);
+		String subject = engine.process(theDetails.getSubjectTemplate(), context);
+
+		SimpleMailMessage email = new SimpleMailMessage();
+		email.setFrom(trim(theDetails.getFrom()));
+		email.setTo(toTrimmedStringArray(theDetails.getTo()));
+		email.setSubject(subject);
+		email.setText(body);
+		email.setSentDate(new Date());
+
+		mySender.send(email);
+
+		ourLog.info("Done sending email (took {}ms)", sw.getMillis());
+	}
+
+	/**
+	 * Set the SMTP server host to use for outbound mail
+	 */
+	public void setSmtpServerHost(String theSmtpServerHost) {
+		mySmtpServerHost = theSmtpServerHost;
+	}
+
+	/**
+	 * Set the SMTP server port to use for outbound mail
+	 */
+	public void setSmtpServerPort(int theSmtpServerPort) {
+		mySmtpServerPort = theSmtpServerPort;
+	}
+
+	private static String[] toTrimmedStringArray(List<String> theTo) {
+		List<String> to = new ArrayList<>();
+		for (String next : theTo) {
+			if (isNotBlank(next)) {
+				to.add(next);
+			}
+		}
+		return to.toArray(new String[to.size()]);
+	}
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/email/IEmailSender.java

@@ -0,0 +1,7 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+public interface IEmailSender {
+
+	void send(EmailDetails theDetails);
+
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/email/SubscriptionDeliveringEmailSubscriber.java

@@ -0,0 +1,87 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+/*-
+ * #%L
+ * HAPI FHIR JPA Server
+ * %%
+ * Copyright (C) 2014 - 2017 University Health Network
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
+import ca.uhn.fhir.jpa.subscription.BaseSubscriptionDeliverySubscriber;
+import ca.uhn.fhir.jpa.subscription.CanonicalSubscription;
+import ca.uhn.fhir.jpa.subscription.ResourceDeliveryMessage;
+import org.apache.commons.lang3.StringUtils;
+import org.hl7.fhir.r4.model.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.commons.lang3.StringUtils.*;
+
+public class SubscriptionDeliveringEmailSubscriber extends BaseSubscriptionDeliverySubscriber {
+	private Logger ourLog = LoggerFactory.getLogger(SubscriptionDeliveringEmailSubscriber.class);
+
+	private SubscriptionEmailInterceptor mySubscriptionEmailInterceptor;
+
+	public SubscriptionDeliveringEmailSubscriber(IFhirResourceDao<?> theSubscriptionDao, Subscription.SubscriptionChannelType theChannelType, SubscriptionEmailInterceptor theSubscriptionEmailInterceptor) {
+		super(theSubscriptionDao, theChannelType, theSubscriptionEmailInterceptor);
+
+		mySubscriptionEmailInterceptor = theSubscriptionEmailInterceptor;
+	}
+
+	@Override
+	public void handleMessage(ResourceDeliveryMessage theMessage) throws Exception {
+		CanonicalSubscription subscription = theMessage.getSubscription();
+
+		// The Subscription.endpoint is treated as the email "to"
+		String endpointUrl = subscription.getEndpointUrl();
+		List<String> destinationAddresses = new ArrayList<>();
+		String[] destinationAddressStrings = StringUtils.split(endpointUrl, ",");
+		for (String next : destinationAddressStrings) {
+			if (isNotBlank(next)) {
+				destinationAddresses.add(trim(next));
+			}
+		}
+
+		String from = defaultString(subscription.getEmailDetails().getFrom(), provideDefaultFrom());
+		String subjectTemplate = defaultString(subscription.getEmailDetails().getSubjectTemplate(), provideDefaultSubjectTemplate());
+		String bodyTemplate = defaultString(subscription.getEmailDetails().getBodyTemplate(), provideDefaultBodyTemplate());
+
+		EmailDetails details = new EmailDetails();
+		details.setTo(destinationAddresses);
+		details.setFrom(from);
+		details.setBodyTemplate(bodyTemplate);
+		details.setSubjectTemplate(subjectTemplate);
+
+		IEmailSender emailSender = mySubscriptionEmailInterceptor.getEmailSender();
+		emailSender.send(details);
+	}
+
+	private String provideDefaultBodyTemplate() {
+		return "A subscription update has been received";
+	}
+
+	private String provideDefaultFrom() {
+		return "unknown@sender.com";
+	}
+
+	private String provideDefaultSubjectTemplate() {
+		return "HAPI FHIR Subscriptions";
+	}
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/email/SubscriptionEmailInterceptor.java

@@ -0,0 +1,68 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+/*-
+ * #%L
+ * HAPI FHIR JPA Server
+ * %%
+ * Copyright (C) 2014 - 2017 University Health Network
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
+import ca.uhn.fhir.jpa.subscription.BaseSubscriptionInterceptor;
+import org.apache.commons.lang3.Validate;
+import org.springframework.beans.factory.annotation.Required;
+
+import javax.annotation.PostConstruct;
+import java.util.List;
+
+public class SubscriptionEmailInterceptor extends BaseSubscriptionInterceptor {
+	private SubscriptionDeliveringEmailSubscriber mySubscriptionDeliverySubscriber;
+	private IEmailSender myEmailSender;
+
+	@Override
+	public org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType getChannelType() {
+		return org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType.EMAIL;
+	}
+
+	public IEmailSender getEmailSender() {
+		return myEmailSender;
+	}
+
+	@Required
+	public void setEmailSender(IEmailSender theEmailSender) {
+		myEmailSender = theEmailSender;
+	}
+
+	@Override
+	protected void registerDeliverySubscriber() {
+		if (mySubscriptionDeliverySubscriber == null) {
+			mySubscriptionDeliverySubscriber = new SubscriptionDeliveringEmailSubscriber(getSubscriptionDao(), getChannelType(), this);
+		}
+		getDeliveryChannel().subscribe(mySubscriptionDeliverySubscriber);
+	}
+
+	@PostConstruct
+	public void start() {
+		Validate.notNull(myEmailSender, "emailSender has not been configured");
+
+		super.start();
+	}
+
+	@Override
+	protected void unregisterDeliverySubscriber() {
+		getDeliveryChannel().unsubscribe(mySubscriptionDeliverySubscriber);
+	}
+}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/resthook/SubscriptionDeliveringRestHookSubscriber.java

@@ -21,10 +21,7 @@ package ca.uhn.fhir.jpa.subscription.resthook;
  */
 
 import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
-import ca.uhn.fhir.jpa.subscription.BaseSubscriptionInterceptor;
+import ca.uhn.fhir.jpa.subscription.*;
-import ca.uhn.fhir.jpa.subscription.BaseSubscriptionSubscriber;
-import ca.uhn.fhir.jpa.subscription.CanonicalSubscription;
-import ca.uhn.fhir.jpa.subscription.ResourceDeliveryMessage;
 import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.EncodingEnum;
 import ca.uhn.fhir.rest.client.api.IGenericClient;
@@ -44,7 +41,7 @@ import java.util.List;
 
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
-public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionSubscriber {
+public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDeliverySubscriber {
 	private Logger ourLog = LoggerFactory.getLogger(SubscriptionDeliveringRestHookSubscriber.class);
 
 	public SubscriptionDeliveringRestHookSubscriber(IFhirResourceDao<?> theSubscriptionDao, Subscription.SubscriptionChannelType theChannelType, BaseSubscriptionInterceptor theSubscriptionInterceptor) {
@@ -78,18 +75,8 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionSu
 	}
 
 	@Override
-	public void handleMessage(Message<?> theMessage) throws MessagingException {
+	public void handleMessage(ResourceDeliveryMessage theMessage) throws MessagingException {
-		if (!(theMessage.getPayload() instanceof ResourceDeliveryMessage)) {
+			CanonicalSubscription subscription = theMessage.getSubscription();
-			return;
-		}
-		try {
-			ResourceDeliveryMessage msg = (ResourceDeliveryMessage) theMessage.getPayload();
-
-			if (!subscriptionTypeApplies(getContext(), msg.getSubscription().getBackingSubscription())) {
-				return;
-			}
-
-			CanonicalSubscription subscription = msg.getSubscription();
 
 			// Grab the endpoint from the subscription
 			String endpointUrl = subscription.getEndpointUrl();
@@ -119,11 +106,7 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionSu
 				}
 			}
 
-			deliverPayload(msg, subscription, payloadType, client);
+			deliverPayload(theMessage, subscription, payloadType, client);
-		} catch (Exception e) {
-			ourLog.error("Failure handling subscription payload", e);
-			throw new MessagingException(theMessage, "Failure handling subscription payload", e);
-		}
 	}
 
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/util/JpaConstants.java

@@ -24,4 +24,8 @@ public class JpaConstants {
 
 	public static final String EXT_SP_UNIQUE = "http://hapifhir.io/fhir/StructureDefinition/sp-unique";
 
+	public static final String EXT_SUBSCRIPTION_EMAIL_FROM = "http://hapifhir.io/fhir/StructureDefinition/subscription-email-from";
+	public static final String EXT_SUBSCRIPTION_SUBJECT_TEMPLATE = "http://hapifhir.io/fhir/StructureDefinition/subscription-email-subject-template";
+	public static final String EXT_SUBSCRIPTION_BODY_TEMPLATE = "http://hapifhir.io/fhir/StructureDefinition/subscription-email-body-template";
+
 }

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

@@ -1,6 +1,7 @@
 package ca.uhn.fhir.jpa.config;
 
 import ca.uhn.fhir.jpa.dao.DaoConfig;
+import ca.uhn.fhir.jpa.subscription.email.SubscriptionEmailInterceptor;
 import ca.uhn.fhir.rest.server.interceptor.RequestValidatingInterceptor;
 import ca.uhn.fhir.validation.ResultSeverityEnum;
 import org.apache.commons.dbcp2.BasicDataSource;

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/dao/BaseJpaTest.java

@@ -8,6 +8,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.sql.SQLException;
 import java.util.*;
+import java.util.concurrent.Callable;
 
 import javax.persistence.EntityManager;
 
@@ -308,4 +309,23 @@ public abstract class BaseJpaTest {
 			fail("Size " + theList.size() + " is != target " + theTarget);
 		}
 	}
+
+	public static void waitForSize(int theTarget, Callable<Number> theCallable) throws Exception {
+		waitForSize(theTarget, 10000, theCallable);
+	}
+
+	public static void waitForSize(int theTarget, int theTimeout, Callable<Number> theCallable) throws Exception {
+		StopWatch sw = new StopWatch();
+		while (theCallable.call().intValue() != theTarget && sw.getMillis() < theTimeout) {
+			try {
+				Thread.sleep(50);
+			} catch (InterruptedException theE) {
+				throw new Error(theE);
+			}
+		}
+		if (sw.getMillis() >= theTimeout) {
+			fail("Size " + theCallable.call() + " is != target " + theTarget);
+		}
+	}
+
 }

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

@@ -0,0 +1,179 @@
+package ca.uhn.fhir.jpa.subscription;
+
+import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
+import ca.uhn.fhir.jpa.provider.BaseResourceProviderDstu2Test;
+import ca.uhn.fhir.jpa.subscription.email.EmailDetails;
+import ca.uhn.fhir.jpa.subscription.email.EmailSender;
+import ca.uhn.fhir.jpa.subscription.email.IEmailSender;
+import ca.uhn.fhir.jpa.subscription.email.SubscriptionEmailInterceptor;
+import ca.uhn.fhir.model.dstu2.composite.CodeableConceptDt;
+import ca.uhn.fhir.model.dstu2.composite.CodingDt;
+import ca.uhn.fhir.model.dstu2.resource.Observation;
+import ca.uhn.fhir.model.dstu2.resource.Subscription;
+import ca.uhn.fhir.model.dstu2.valueset.ObservationStatusEnum;
+import ca.uhn.fhir.model.dstu2.valueset.SubscriptionChannelTypeEnum;
+import ca.uhn.fhir.model.dstu2.valueset.SubscriptionStatusEnum;
+import ca.uhn.fhir.rest.api.MethodOutcome;
+import com.icegreen.greenmail.imap.ImapConstants;
+import com.icegreen.greenmail.store.MailFolder;
+import com.icegreen.greenmail.store.Store;
+import com.icegreen.greenmail.util.GreenMail;
+import com.icegreen.greenmail.util.GreenMailUtil;
+import com.icegreen.greenmail.util.ServerSetupTest;
+import org.hl7.fhir.instance.model.api.IIdType;
+import org.junit.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.mail.internet.InternetAddress;
+import javax.mail.internet.MimeMessage;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Callable;
+
+import static org.junit.Assert.*;
+
+public class EmailSubscriptionDstu2Test extends BaseResourceProviderDstu2Test {
+
+	private static final Logger ourLog = LoggerFactory.getLogger(EmailSubscriptionDstu2Test.class);
+	private static GreenMail ourTestSmtp;
+	private SubscriptionEmailInterceptor mySubscriber;
+	private List<IIdType> mySubscriptionIds = new ArrayList<>();
+
+	@Autowired
+	private List<IFhirResourceDao<?>> myResourceDaos;
+
+	@After
+	public void after() throws Exception {
+		ourLog.info("** AFTER **");
+		super.after();
+
+		for (IIdType next : mySubscriptionIds) {
+			ourClient.delete().resourceById(next).execute();
+		}
+		mySubscriptionIds.clear();
+
+		ourRestServer.unregisterInterceptor(mySubscriber);
+	}
+
+	@Before
+	public void before() throws Exception {
+		super.before();
+
+		EmailSender emailSender = new EmailSender();
+		emailSender.setSmtpServerHost("localhost");
+		emailSender.setSmtpServerPort(3025);
+		emailSender.start();
+
+		mySubscriber = new SubscriptionEmailInterceptor();
+		mySubscriber.setEmailSender(emailSender);
+		mySubscriber.setResourceDaos(myResourceDaos);
+		mySubscriber.setFhirContext(myFhirCtx);
+		mySubscriber.start();
+		ourRestServer.registerInterceptor(mySubscriber);
+
+//		ourLog.info("Sending test email to warm up the server");
+//		EmailDetails details = new EmailDetails();
+//		details.setFrom("a@a.com");
+//		details.setTo(Arrays.asList("b@b.com"));
+//		details.setSubjectTemplate("SUBJ");
+//		details.setBodyTemplate("BODY");
+//		emailSender.send(details);
+//		ourLog.info("Done sending test email to warm up the server");
+//		Store store = ourTestSmtp.getManagers().getImapHostManager().getStore();
+//		MailFolder mailbox = store.getMailbox(ImapConstants.USER_NAMESPACE);
+//		mailbox.deleteAllMessages();
+	}
+
+	private Subscription createSubscription(String criteria, String payload, String endpoint) throws InterruptedException {
+		Subscription subscription = new Subscription();
+		subscription.setReason("Monitor new neonatal function (note, age will be determined by the monitor)");
+		subscription.setStatus(SubscriptionStatusEnum.REQUESTED);
+		subscription.setCriteria(criteria);
+
+		Subscription.Channel channel = new Subscription.Channel();
+		channel.setType(SubscriptionChannelTypeEnum.EMAIL);
+		channel.setPayload(payload);
+		channel.setEndpoint(endpoint);
+		subscription.setChannel(channel);
+
+		MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
+		subscription.setId(methodOutcome.getId().getIdPart());
+		mySubscriptionIds.add(methodOutcome.getId());
+
+		RestHookTestDstu2Test.waitForQueueToDrain(ourRestHookSubscriptionInterceptor);
+
+		return subscription;
+	}
+
+	private Observation sendObservation(String code, String system) {
+		Observation observation = new Observation();
+		CodeableConceptDt codeableConcept = new CodeableConceptDt();
+		observation.setCode(codeableConcept);
+		CodingDt coding = codeableConcept.addCoding();
+		coding.setCode(code);
+		coding.setSystem(system);
+
+		observation.setStatus(ObservationStatusEnum.FINAL);
+
+		MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
+
+		String observationId = methodOutcome.getId().getIdPart();
+		observation.setId(observationId);
+
+		return observation;
+	}
+
+	@Test
+	public void testSubscribeAndDeliver() throws Exception {
+		String payload = "application/json";
+		String code = "1000000050";
+		String criteria1 = "Observation?code=SNOMED-CT|" + code + "&_format=xml";
+
+		Subscription subscription1 = createSubscription(criteria1, payload, "to1@example.com,to2@example.com");
+		RestHookTestDstu2Test.waitForQueueToDrain(ourRestHookSubscriptionInterceptor);
+
+		assertEquals(0, Arrays.asList(ourTestSmtp.getReceivedMessages()).size());
+
+		Observation observation1 = sendObservation(code, "SNOMED-CT");
+
+		waitForSize(2, 20000, new Callable<Number>(){
+			@Override
+			public Number call() throws Exception {
+				return ourTestSmtp.getReceivedMessages().length;
+			}
+		});
+
+		MimeMessage[] messages = ourTestSmtp.getReceivedMessages();
+		assertEquals(2, messages.length);
+		int msgIdx = 0;
+		ourLog.info("Received: " + GreenMailUtil.getWholeMessage(messages[msgIdx]));
+		assertEquals("HAPI FHIR Subscriptions", messages[msgIdx].getSubject());
+		assertEquals(1, messages[msgIdx].getFrom().length);
+		assertEquals("unknown@sender.com", ((InternetAddress) messages[msgIdx].getFrom()[0]).getAddress());
+		assertEquals(2, messages[msgIdx].getAllRecipients().length);
+		assertEquals("to1@example.com", ((InternetAddress) messages[msgIdx].getAllRecipients()[0]).getAddress());
+		assertEquals("to2@example.com", ((InternetAddress) messages[msgIdx].getAllRecipients()[1]).getAddress());
+		assertEquals(1, messages[msgIdx].getHeader("Content-Type").length);
+		assertEquals("text/plain; charset=UTF-8", messages[msgIdx].getHeader("Content-Type")[0]);
+		String foundBody = GreenMailUtil.getBody(messages[msgIdx]);
+		assertEquals("A subscription update has been received", foundBody);
+
+	}
+
+
+	@AfterClass
+	public static void afterClass() {
+		ourTestSmtp.stop();
+	}
+
+	@BeforeClass
+	public static void beforeClass() {
+		ourTestSmtp = new GreenMail(ServerSetupTest.SMTP);
+		ourTestSmtp.start();
+	}
+
+
+}

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

@@ -278,12 +278,15 @@ public class RestHookTestDstu2Test extends BaseResourceProviderDstu2Test {
 		Assert.assertFalse(observation2.getId().isEmpty());
 	}
 
-	public static void waitForQueueToDrain(BaseSubscriptionInterceptor theRestHookSubscriptionInterceptor) throws InterruptedException {
+	public static void waitForQueueToDrain(BaseSubscriptionInterceptor theSubscriptionInterceptor) throws InterruptedException {
-		ourLog.info("QUEUE HAS {} ITEMS", theRestHookSubscriptionInterceptor.getExecutorQueueSizeForUnitTests());
+		Thread.sleep(50);
-		while (theRestHookSubscriptionInterceptor.getExecutorQueueSizeForUnitTests() > 0) {
+		ourLog.info("Executor work queue has {} items", theSubscriptionInterceptor.getExecutorQueueSizeForUnitTests());
+		if (theSubscriptionInterceptor.getExecutorQueueSizeForUnitTests() > 0) {
+			while (theSubscriptionInterceptor.getExecutorQueueSizeForUnitTests() > 0) {
 				Thread.sleep(50);
 			}
-		ourLog.info("QUEUE HAS {} ITEMS", theRestHookSubscriptionInterceptor.getExecutorQueueSizeForUnitTests());
+			ourLog.info("Executor work queue has {} items", theSubscriptionInterceptor.getExecutorQueueSizeForUnitTests());
+		}
 	}
 
 	private void waitForQueueToDrain() throws InterruptedException {

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

@@ -46,6 +46,8 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 
 	@After
 	public void afterUnregisterRestHookListener() {
+		ourLog.info("**** Starting @After *****");
+
 		for (IIdType next : mySubscriptionIds){
 			ourClient.delete().resourceById(next).execute();
 		}
@@ -220,6 +222,7 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		Subscription subscription1 = createSubscription(criteria1, payload, ourListenerServerBase);
 		Subscription subscription2 = createSubscription(criteria2, payload, ourListenerServerBase);
 
+		ourLog.info("About to send observation 1");
 		Observation observation1 = sendObservation(code, "SNOMED-CT");
 
 		// Should see 1 subscription notification
@@ -236,6 +239,7 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		waitForQueueToDrain();
 
 		// Send another observation
+		ourLog.info("About to send observation 2");
 		Observation observation2 = sendObservation(code, "SNOMED-CT");
 
 		// Should see two subscription notifications
@@ -247,6 +251,7 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 		waitForQueueToDrain();
 
 		// Send another
+		ourLog.info("About to send observation 3");
 		Observation observationTemp3 = sendObservation(code, "SNOMED-CT");
 
 		// Should see only one subscription notification
@@ -370,9 +375,9 @@ public class RestHookTestDstu3Test extends BaseResourceProviderDstu3Test {
 
 		@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(";.*", ""));
+			ourLog.info("Received Listener Update (now have {} updates)", ourUpdatedObservations.size());
 			return new MethodOutcome(new IdType("Observation/1"), false);
 		}
 

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

@@ -0,0 +1,65 @@
+package ca.uhn.fhir.jpa.subscription.email;
+
+import com.icegreen.greenmail.util.GreenMail;
+import com.icegreen.greenmail.util.GreenMailUtil;
+import com.icegreen.greenmail.util.ServerSetupTest;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.mail.internet.InternetAddress;
+import javax.mail.internet.MimeMessage;
+import java.util.Arrays;
+
+import static org.junit.Assert.*;
+
+public class EmailSenderTest {
+
+	private static final Logger ourLog = LoggerFactory.getLogger(EmailSenderTest.class);
+	private static GreenMail ourTestSmtp;
+
+	@Test
+	public void testSend() throws Exception {
+		EmailSender sender = new EmailSender();
+		sender.setSmtpServerHost("localhost");
+		sender.setSmtpServerPort(3025);
+		sender.start();
+
+		String body = "foo";
+
+		EmailDetails details = new EmailDetails();
+		details.setFrom("foo@example.com ");
+		details.setTo(Arrays.asList(" to1@example.com", "to2@example.com   "));
+		details.setSubjectTemplate("test subject");
+		details.setBodyTemplate(body);
+		sender.send(details);
+
+		MimeMessage[] messages = ourTestSmtp.getReceivedMessages();
+		assertEquals(2, messages.length);
+		ourLog.info("Received: " + GreenMailUtil.getWholeMessage(messages[0]));
+		assertEquals("test subject", messages[0].getSubject());
+		assertEquals(1, messages[0].getFrom().length);
+		assertEquals("foo@example.com", ((InternetAddress)messages[0].getFrom()[0]).getAddress());
+		assertEquals(2, messages[0].getAllRecipients().length);
+		assertEquals("to1@example.com", ((InternetAddress)messages[0].getAllRecipients()[0]).getAddress());
+		assertEquals("to2@example.com", ((InternetAddress)messages[0].getAllRecipients()[1]).getAddress());
+		assertEquals(1, messages[0].getHeader("Content-Type").length);
+		assertEquals("text/plain; charset=UTF-8", messages[0].getHeader("Content-Type")[0]);
+		String foundBody = GreenMailUtil.getBody(messages[0]);
+		assertEquals("foo", foundBody);
+	}
+
+	@AfterClass
+	public static void afterClass() {
+		ourTestSmtp.stop();
+	}
+
+	@BeforeClass
+	public static void beforeClass() {
+		ourTestSmtp = new GreenMail(ServerSetupTest.SMTP);
+		ourTestSmtp.start();
+	}
+
+}

pom.xml

@@ -422,6 +422,16 @@
 				<artifactId>commonmark</artifactId>
 				<version>0.9.0</version>
 			</dependency>
+			<dependency>
+				<groupId>com.icegreen</groupId>
+				<artifactId>greenmail</artifactId>
+				<version>1.5.5</version>
+			</dependency>
+			<dependency>
+				<groupId>com.icegreen</groupId>
+				<artifactId>greenmail-spring</artifactId>
+				<version>1.5.5</version>
+			</dependency>
 			<dependency>
 				<groupId>com.github.bkiers</groupId>
 				<artifactId>Liqp</artifactId>
@@ -457,6 +467,11 @@
 				<artifactId>okhttp</artifactId>
 				<version>3.8.1</version>
 			</dependency>
+			<dependency>
+				<groupId>com.sun.mail</groupId>
+				<artifactId>javax.mail</artifactId>
+				<version>1.6.0</version>
+			</dependency>
 			<dependency>
 				<groupId>commons-cli</groupId>
 				<artifactId>commons-cli</artifactId>