fixed the issue of meta.source field inconsistently populated in subscription messages for different requests (#4524)

* fix + test

* minor fix

* Addressing suggestion

* Minor changes

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/6_4_0/4523-meta-source-field-in-subscription-messages-is-inconsistently-populated.yaml

@@ -0,0 +1,5 @@
+---
+type: fix
+issue: 4523
+title: "Previously, meta source field in subscription messages was inconsistently populated regarding different requests.
+Now, this has been fixed and meta source will be included in all subscription messages."

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

@@ -34,7 +34,9 @@ import ca.uhn.fhir.jpa.subscription.match.registry.ActiveSubscription;
 import ca.uhn.fhir.jpa.subscription.match.registry.SubscriptionRegistry;
 import ca.uhn.fhir.jpa.subscription.model.CanonicalSubscription;
 import ca.uhn.fhir.jpa.subscription.model.ResourceDeliveryMessage;
+import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
+import ca.uhn.fhir.rest.api.server.SystemRequestDetails;
 import ca.uhn.fhir.util.BundleBuilder;
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.text.StringSubstitutor;
@@ -139,6 +141,14 @@ public abstract class BaseSubscriptionDeliverySubscriber implements MessageHandl
 		return builder.getBundle();
 	}
 
+	protected IBaseResource updateDeliveryResourceWithMetaSource(IBaseResource thePayloadResource) {
+		String resType = thePayloadResource.fhirType();
+		IFhirResourceDao<?> dao = myDaoRegistry.getResourceDao(resType);
+		IBaseResource resourceWithMetaSource = dao.read(thePayloadResource.getIdElement(), new SystemRequestDetails());
+
+		return resourceWithMetaSource;
+	}
+
 	@VisibleForTesting
 	public void setFhirContextForUnitTest(FhirContext theCtx) {
 		myFhirContext = theCtx;

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/match/deliver/message/SubscriptionDeliveringMessageSubscriber.java

@@ -32,6 +32,7 @@ import ca.uhn.fhir.jpa.subscription.model.ResourceDeliveryMessage;
 import ca.uhn.fhir.jpa.subscription.model.ResourceModifiedJsonMessage;
 import ca.uhn.fhir.jpa.subscription.model.ResourceModifiedMessage;
 import ca.uhn.fhir.rest.api.EncodingEnum;
+import ca.uhn.fhir.util.HapiExtensions;
 import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -59,8 +60,14 @@ public class SubscriptionDeliveringMessageSubscriber extends BaseSubscriptionDel
 
 	protected void doDelivery(ResourceDeliveryMessage theSourceMessage, CanonicalSubscription theSubscription, IChannelProducer theChannelProducer, ResourceModifiedJsonMessage theWrappedMessageToSend) {
 		String payloadId = theSourceMessage.getPayloadId();
+		IBaseResource payloadResource = null;
 		if (isNotBlank(theSubscription.getPayloadSearchCriteria())) {
-			IBaseResource payloadResource = createDeliveryBundleForPayloadSearchCriteria(theSubscription, theWrappedMessageToSend.getPayload().getPayload(myFhirContext));
+			payloadResource = createDeliveryBundleForPayloadSearchCriteria(theSubscription, theWrappedMessageToSend.getPayload().getPayload(myFhirContext));
+		} else if (! theWrappedMessageToSend.getPayload().getPayloadString().contains(HapiExtensions.EXT_META_SOURCE)){
+			payloadResource = updateDeliveryResourceWithMetaSource(theWrappedMessageToSend.getPayload().getPayload(myFhirContext));
+		}
+
+		if (payloadResource != null) {
 			ResourceModifiedJsonMessage newWrappedMessageToSend = convertDeliveryMessageToResourceModifiedMessage(theSourceMessage, payloadResource);
 			theWrappedMessageToSend.setPayload(newWrappedMessageToSend.getPayload());
 			payloadId = payloadResource.getIdElement().toUnqualifiedVersionless().getValue();

hapi-fhir-jpaserver-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/match/deliver/BaseSubscriptionDeliverySubscriberTest.java

@@ -27,7 +27,6 @@ import ca.uhn.fhir.rest.client.api.IRestfulClientFactory;
 import ca.uhn.fhir.rest.server.SimpleBundleProvider;
 import ca.uhn.fhir.rest.server.exceptions.InternalErrorException;
 import com.fasterxml.jackson.core.JsonProcessingException;
-import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.r4.model.Bundle;
 import org.hl7.fhir.r4.model.IdType;
 import org.hl7.fhir.r4.model.Patient;
@@ -59,6 +58,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.eq;
@@ -208,6 +208,7 @@ public class BaseSubscriptionDeliverySubscriberTest {
 		});
 		when(myInterceptorBroadcaster.callHooks(eq(Pointcut.SUBSCRIPTION_AFTER_MESSAGE_DELIVERY), any())).thenReturn(false);
 		when(myChannelFactory.getOrCreateProducer(any(), any(), any())).thenReturn(myChannelProducer);
+		when(myDaoRegistry.getResourceDao(anyString())).thenReturn(myResourceDao);
 
 		CanonicalSubscription subscription = generateSubscription();
 		Patient patient = generatePatient();
@@ -345,6 +346,7 @@ public class BaseSubscriptionDeliverySubscriberTest {
 		when(myInterceptorBroadcaster.callHooks(eq(Pointcut.SUBSCRIPTION_BEFORE_MESSAGE_DELIVERY), any())).thenReturn(true);
 		when(myInterceptorBroadcaster.callHooks(eq(Pointcut.SUBSCRIPTION_AFTER_MESSAGE_DELIVERY), any())).thenReturn(false);
 		when(myChannelFactory.getOrCreateProducer(any(), any(), any())).thenReturn(myChannelProducer);
+		when(myDaoRegistry.getResourceDao(anyString())).thenReturn(myResourceDao);
 
 		CanonicalSubscription subscription = generateSubscription();
 		Patient patient = generatePatient();
@@ -401,6 +403,40 @@ public class BaseSubscriptionDeliverySubscriberTest {
 		}
 	}
 
+	@Test
+	public void testSubscriptionMessageContainsMetaSourceField() throws URISyntaxException {
+		//Given: we have a subscription message that contains a patient resource
+		Patient p1 = generatePatient();
+		p1.addName().setFamily("p1-family");
+
+		CanonicalSubscription subscription = generateSubscription();
+		subscription.setCriteriaString("[*]");
+
+		ResourceDeliveryMessage payload = new ResourceDeliveryMessage();
+		payload.setSubscription(subscription);
+		payload.setPayload(myCtx, p1, EncodingEnum.JSON);
+		payload.setOperationType(ResourceModifiedMessage.OperationTypeEnum.CREATE);
+
+		//When
+		when(myInterceptorBroadcaster.callHooks(eq(Pointcut.SUBSCRIPTION_BEFORE_MESSAGE_DELIVERY), ArgumentMatchers.any(HookParams.class))).thenReturn(true);
+		when(myInterceptorBroadcaster.callHooks(eq(Pointcut.SUBSCRIPTION_AFTER_MESSAGE_DELIVERY), any())).thenReturn(false);
+		when(myChannelFactory.getOrCreateProducer(any(), any(), any())).thenReturn(myChannelProducer);
+		when(myDaoRegistry.getResourceDao(anyString())).thenReturn(myResourceDao);
+
+		p1.getMeta().setSource("#example-source");
+		when(myResourceDao.read(any(), any())).thenReturn(p1);
+
+		//Then: meta.source field will be included in the message
+		myMessageSubscriber.handleMessage(payload);
+
+		ArgumentCaptor<ResourceModifiedJsonMessage> captor = ArgumentCaptor.forClass(ResourceModifiedJsonMessage.class);
+		verify(myChannelProducer).send(captor.capture());
+		List<ResourceModifiedJsonMessage> messages = captor.getAllValues();
+
+		ResourceModifiedMessage receivedMessage = messages.get(0).getPayload();
+		assertTrue(receivedMessage.getPayloadString().contains("\"source\":\"#example-source\""));
+	}
+
 	@Nonnull
 	private Patient generatePatient() {
 		Patient patient = new Patient();