Work on subscriptions

This commit is contained in:
jamesagnew 2020-06-29 09:39:10 -04:00
parent 52addcff59
commit ed2e62752e
10 changed files with 290 additions and 78 deletions

View File

@ -0,0 +1,110 @@
package ca.uhn.fhir.util;
import ca.uhn.fhir.context.BaseRuntimeChildDefinition;
import ca.uhn.fhir.context.BaseRuntimeElementDefinition;
import ca.uhn.fhir.context.FhirContext;
import ca.uhn.fhir.context.RuntimeResourceDefinition;
import org.hl7.fhir.instance.model.api.IBase;
import org.hl7.fhir.instance.model.api.IBaseBundle;
import org.hl7.fhir.instance.model.api.IBaseResource;
import org.hl7.fhir.instance.model.api.IPrimitiveType;
import org.thymeleaf.util.Validate;
/**
* This class can be used to build a Bundle resource to be used as a FHIR transaction.
*
* This is not yet complete, and doesn't support all FHIR features. <b>USE WITH CAUTION</b> as the API
* may change.
*
* @since 5.1.0
*/
public class TransactionBuilder {
private final FhirContext myContext;
private final IBaseBundle myBundle;
private final RuntimeResourceDefinition myBundleDef;
private final BaseRuntimeChildDefinition myEntryChild;
private final BaseRuntimeElementDefinition<?> myEntryDef;
private final BaseRuntimeChildDefinition myEntryResourceChild;
private final BaseRuntimeChildDefinition myEntryFullUrlChild;
private final BaseRuntimeChildDefinition myEntryRequestChild;
private final BaseRuntimeElementDefinition<?> myEntryRequestDef;
private final BaseRuntimeChildDefinition myEntryRequestUrlChild;
private final BaseRuntimeChildDefinition myEntryRequestMethodChild;
private final BaseRuntimeElementDefinition<?> myEntryRequestMethodDef;
/**
* Constructor
*/
public TransactionBuilder(FhirContext theContext) {
myContext = theContext;
myBundleDef = myContext.getResourceDefinition("Bundle");
myBundle = (IBaseBundle) myBundleDef.newInstance();
BaseRuntimeChildDefinition typeChild = myBundleDef.getChildByName("type");
IPrimitiveType<?> type = (IPrimitiveType<?>) typeChild.getChildByName("type").newInstance(typeChild.getInstanceConstructorArguments());
type.setValueAsString("transaction");
typeChild.getMutator().setValue(myBundle, type);
myEntryChild = myBundleDef.getChildByName("entry");
myEntryDef = myEntryChild.getChildByName("entry");
myEntryResourceChild = myEntryDef.getChildByName("resource");
myEntryFullUrlChild = myEntryDef.getChildByName("fullUrl");
myEntryRequestChild = myEntryDef.getChildByName("request");
myEntryRequestDef = myEntryRequestChild.getChildByName("request");
myEntryRequestUrlChild = myEntryRequestDef.getChildByName("url");
myEntryRequestMethodChild = myEntryRequestDef.getChildByName("method");
myEntryRequestMethodDef = myEntryRequestMethodChild.getChildByName("method");
}
/**
* Adds an entry containing an update (PUT) request
*
* @param theResource The resource to update
*/
public TransactionBuilder addUpdateEntry(IBaseResource theResource) {
Validate.notNull(theResource, "theResource must not be null");
Validate.notEmpty(theResource.getIdElement().getValue(), "theResource must have an ID");
IBase entry = myEntryDef.newInstance();
myEntryChild.getMutator().addValue(myBundle, entry);
// Bundle.entry.fullUrl
IPrimitiveType<?> fullUrl = (IPrimitiveType<?>) myContext.getElementDefinition("uri").newInstance();
fullUrl.setValueAsString(theResource.getIdElement().getValue());
myEntryFullUrlChild.getMutator().setValue(entry, fullUrl);
// Bundle.entry.resource
myEntryResourceChild.getMutator().setValue(entry, theResource);
// Bundle.entry.request
IBase request = myEntryRequestDef.newInstance();
myEntryRequestChild.getMutator().setValue(entry, request);
// Bundle.entry.request.url
IPrimitiveType<?> url = (IPrimitiveType<?>) myContext.getElementDefinition("uri").newInstance();
url.setValueAsString(theResource.getIdElement().toUnqualifiedVersionless().getValue());
myEntryRequestUrlChild.getMutator().setValue(request, url);
// Bundle.entry.request.url
IPrimitiveType<?> method = (IPrimitiveType<?>) myEntryRequestMethodDef.newInstance(myEntryRequestMethodChild.getInstanceConstructorArguments());
method.setValueAsString("PUT");
myEntryRequestMethodChild.getMutator().setValue(request, method);
return this;
}
public IBaseBundle getBundle() {
return myBundle;
}
}

View File

@ -8,6 +8,7 @@ import ca.uhn.fhir.jpa.subscription.submit.interceptor.SubscriptionMatcherInterc
import ca.uhn.fhir.rest.annotation.Create;
import ca.uhn.fhir.rest.annotation.ResourceParam;
import ca.uhn.fhir.rest.annotation.Transaction;
import ca.uhn.fhir.rest.annotation.TransactionParam;
import ca.uhn.fhir.rest.annotation.Update;
import ca.uhn.fhir.rest.api.Constants;
import ca.uhn.fhir.rest.api.MethodOutcome;
@ -211,7 +212,8 @@ public abstract class BaseSubscriptionsR4Test extends BaseResourceProviderR4Test
public static class PlainProvider {
@Transaction
public Bundle transaction(@RequestParam Bundle theInput) {
public Bundle transaction(@TransactionParam Bundle theInput) {
ourLog.info("Received transaction update");
ourTransactions.add(theInput);
return theInput;
}
@ -231,8 +233,8 @@ public abstract class BaseSubscriptionsR4Test extends BaseResourceProviderR4Test
public static void startListenerServer() throws Exception {
RestfulServer ourListenerRestServer = new RestfulServer(FhirContext.forR4());
ObservationResourceProvider observationResourceProvider = new ObservationResourceProvider();
ourListenerRestServer.setResourceProviders(observationResourceProvider);
ourListenerRestServer.registerProvider(new ObservationResourceProvider());
ourListenerRestServer.registerProvider(new PlainProvider());
ourListenerServer = new Server(0);

View File

@ -1014,10 +1014,10 @@ public class RestHookTestR4Test extends BaseSubscriptionsR4Test {
@Test
public void testDeliverSearchSet() throws Exception {
public void testDeliverSearchResult() throws Exception {
{
Subscription subscription = newSubscription("Observation?", "application/json");
subscription.addExtension(JpaConstants.EXT_SUBSCRIPTION_DELIVER_BUNDLE_SEARCH_RESULT, new StringType("Observation?_id=${resource_id}&_include=*"));
subscription.addExtension(JpaConstants.EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT, new StringType("Observation?_id=${matched_resource_id}&_include=*"));
MethodOutcome methodOutcome = ourClient.create().resource(subscription).execute();
mySubscriptionIds.add(methodOutcome.getId());
waitForActivatedSubscriptionCount(1);
@ -1030,11 +1030,17 @@ public class RestHookTestR4Test extends BaseSubscriptionsR4Test {
Observation observation = new Observation();
observation.addExtension().setUrl("Observation#accessType").setValue(new Coding().setCode("Catheter"));
observation.getSubject().setReferenceElement(patientId);
observation.getSubject().setReferenceElement(patientId.toUnqualifiedVersionless());
MethodOutcome methodOutcome = ourClient.create().resource(observation).execute();
assertEquals(true, methodOutcome.getCreated());
waitForQueueToDrain();
waitForSize(1, ourTransactions);
ourLog.info("Received transaction: {}", myFhirCtx.newJsonParser().setPrettyPrint(true).encodeResourceToString(ourTransactions.get(0)));
Bundle xact = ourTransactions.get(0);
assertEquals(2, xact.getEntry().size());
}
}

View File

@ -215,9 +215,9 @@ public class JpaConstants {
*/
public static final String EXT_EXTERNALIZED_BINARY_ID = "http://hapifhir.io/fhir/StructureDefinition/externalized-binary-id";
/**
* For subscription, deliver a bundle containinf a search result instead of just a single resource
* For subscription, deliver a bundle containing a search result instead of just a single resource
*/
public static final String EXT_SUBSCRIPTION_DELIVER_BUNDLE_SEARCH_RESULT = "http://hapifhir.io/fhir/StructureDefinition/subscription-deliver-bundle-search-result";
public static final String EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT = "http://hapifhir.io/fhir/StructureDefinition/subscription-payload-search-result";
/**
* Placed in system-generated extensions
*/

View File

@ -74,7 +74,7 @@ public class SubscriptionDeliveringMessageSubscriber extends BaseSubscriptionDel
public void handleMessage(ResourceDeliveryMessage theMessage) throws MessagingException, URISyntaxException {
CanonicalSubscription subscription = theMessage.getSubscription();
// Interceptor call: SUBSCRIPTION_BEFORE_REST_HOOK_DELIVERY
// Interceptor call: SUBSCRIPTION_BEFORE_MESSAGE_DELIVERY
HookParams params = new HookParams()
.add(CanonicalSubscription.class, subscription)
.add(ResourceDeliveryMessage.class, theMessage);
@ -104,7 +104,7 @@ public class SubscriptionDeliveringMessageSubscriber extends BaseSubscriptionDel
deliverPayload(theMessage, subscription, channelProducer);
// Interceptor call: SUBSCRIPTION_AFTER_REST_HOOK_DELIVERY
// Interceptor call: SUBSCRIPTION_AFTER_MESSAGE_DELIVERY
params = new HookParams()
.add(CanonicalSubscription.class, subscription)
.add(ResourceDeliveryMessage.class, theMessage);

View File

@ -25,11 +25,14 @@ import ca.uhn.fhir.interceptor.api.HookParams;
import ca.uhn.fhir.interceptor.api.Pointcut;
import ca.uhn.fhir.jpa.api.dao.DaoRegistry;
import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao;
import ca.uhn.fhir.jpa.searchparam.MatchUrlService;
import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
import ca.uhn.fhir.jpa.subscription.match.deliver.BaseSubscriptionDeliverySubscriber;
import ca.uhn.fhir.jpa.subscription.model.CanonicalSubscription;
import ca.uhn.fhir.jpa.subscription.model.ResourceDeliveryMessage;
import ca.uhn.fhir.rest.api.EncodingEnum;
import ca.uhn.fhir.rest.api.RequestTypeEnum;
import ca.uhn.fhir.rest.api.server.IBundleProvider;
import ca.uhn.fhir.rest.client.api.Header;
import ca.uhn.fhir.rest.client.api.IGenericClient;
import ca.uhn.fhir.rest.client.api.IHttpClient;
@ -40,7 +43,10 @@ import ca.uhn.fhir.rest.client.interceptor.SimpleRequestHeaderInterceptor;
import ca.uhn.fhir.rest.gclient.IClientExecutable;
import ca.uhn.fhir.rest.server.exceptions.ResourceGoneException;
import ca.uhn.fhir.rest.server.exceptions.ResourceNotFoundException;
import ca.uhn.fhir.util.TransactionBuilder;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.text.StringSubstitutor;
import org.apache.http.NameValuePair;
import org.hl7.fhir.instance.model.api.IBaseResource;
import org.hl7.fhir.instance.model.api.IIdType;
import org.slf4j.Logger;
@ -65,6 +71,9 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
@Autowired
private DaoRegistry myDaoRegistry;
@Autowired
private MatchUrlService myMatchUrlService;
/**
* Constructor
*/
@ -81,6 +90,30 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
protected void doDelivery(ResourceDeliveryMessage theMsg, CanonicalSubscription theSubscription, EncodingEnum thePayloadType, IGenericClient theClient, IBaseResource thePayloadResource) {
IClientExecutable<?, ?> operation;
if (isNotBlank(theSubscription.getPayloadSearchResult())) {
String resType = theSubscription.getPayloadSearchResult().substring(0, theSubscription.getPayloadSearchResult().indexOf('?'));
IFhirResourceDao<?> dao = myDaoRegistry.getResourceDao(resType);
RuntimeResourceDefinition resourceDefinition = myFhirContext.getResourceDefinition(resType);
String payloadUrl = theSubscription.getPayloadSearchResult();
Map<String, String> valueMap = new HashMap<>(1);
valueMap.put("matched_resource_id", thePayloadResource.getIdElement().toUnqualifiedVersionless().getValue());
payloadUrl = new StringSubstitutor(valueMap).replace(payloadUrl);
SearchParameterMap payloadSearchMap = myMatchUrlService.translateMatchUrl(payloadUrl, resourceDefinition);
payloadSearchMap.setLoadSynchronous(true);
IBundleProvider searchResults = dao.search(payloadSearchMap);
TransactionBuilder builder = new TransactionBuilder(myFhirContext);
for (IBaseResource next : searchResults.getResources(0, searchResults.size())) {
builder.addUpdateEntry(next);
}
operation = theClient.transaction().withBundle(builder.getBundle());
} else {
switch (theMsg.getOperationType()) {
case CREATE:
case UPDATE:
@ -108,6 +141,8 @@ public class SubscriptionDeliveringRestHookSubscriber extends BaseSubscriptionDe
return;
}
}
if (thePayloadType != null) {
operation.encoded(thePayloadType);
}

View File

@ -113,6 +113,7 @@ public class SubscriptionCanonicalizer {
retVal.setChannelExtensions(extractExtension(subscription));
retVal.setIdElement(subscription.getIdElement());
retVal.setPayloadString(subscription.getChannel().getPayload());
retVal.setPayloadSearchResult(getExtensionString(subscription, JpaConstants.EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT));
if (retVal.getChannelType() == CanonicalSubscriptionChannelType.EMAIL) {
String from;
@ -208,6 +209,7 @@ public class SubscriptionCanonicalizer {
retVal.setChannelExtensions(extractExtension(subscription));
retVal.setIdElement(subscription.getIdElement());
retVal.setPayloadString(subscription.getChannel().getPayload());
retVal.setPayloadSearchResult(getExtensionString(subscription, JpaConstants.EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT));
if (retVal.getChannelType() == CanonicalSubscriptionChannelType.EMAIL) {
String from;
@ -261,7 +263,7 @@ public class SubscriptionCanonicalizer {
retVal.setChannelExtensions(extractExtension(subscription));
retVal.setIdElement(subscription.getIdElement());
retVal.setPayloadString(subscription.getContentType());
retVal.setDeliverBundleSearchResult(getExtensionString(subscription, JpaConstants.EXT_SUBSCRIPTION_DELIVER_BUNDLE_SEARCH_RESULT));
retVal.setPayloadSearchResult(getExtensionString(subscription, JpaConstants.EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT));
if (retVal.getChannelType() == CanonicalSubscriptionChannelType.EMAIL) {
String from;

View File

@ -34,7 +34,11 @@ import org.hl7.fhir.r4.model.Subscription;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
@ -64,8 +68,8 @@ public class CanonicalSubscription implements Serializable, Cloneable, IModelJso
private RestHookDetails myRestHookDetails;
@JsonProperty("extensions")
private Map<String, List<String>> myChannelExtensions;
@JsonProperty("deliverBundleSearchResult")
private String myDeliverBundleSearchResult;
@JsonProperty("payloadSearchResult")
private String myPayloadSearchResult;
/**
* Constructor
@ -74,6 +78,14 @@ public class CanonicalSubscription implements Serializable, Cloneable, IModelJso
super();
}
public String getPayloadSearchResult() {
return myPayloadSearchResult;
}
public void setPayloadSearchResult(String thePayloadSearchResult) {
myPayloadSearchResult = thePayloadSearchResult;
}
/**
* For now we're using the R4 TriggerDefinition, but this
* may change in the future when things stabilize
@ -82,7 +94,6 @@ public class CanonicalSubscription implements Serializable, Cloneable, IModelJso
myTrigger = theTrigger;
}
public CanonicalSubscriptionChannelType getChannelType() {
return myChannelType;
}
@ -278,8 +289,21 @@ public class CanonicalSubscription implements Serializable, Cloneable, IModelJso
}
}
public void setDeliverBundleSearchResult(String theDeliverBundleSearchResult) {
myDeliverBundleSearchResult = theDeliverBundleSearchResult;
@Override
public String toString() {
return new ToStringBuilder(this)
.append("myIdElement", myIdElement)
.append("myStatus", myStatus)
.append("myCriteriaString", myCriteriaString)
.append("myEndpointUrl", myEndpointUrl)
.append("myPayloadString", myPayloadString)
// .append("myHeaders", myHeaders)
.append("myChannelType", myChannelType)
// .append("myTrigger", myTrigger)
// .append("myEmailDetails", myEmailDetails)
// .append("myRestHookDetails", myRestHookDetails)
// .append("myChannelExtensions", myChannelExtensions)
.toString();
}
public static class EmailDetails implements IModelJson {
@ -400,21 +424,4 @@ public class CanonicalSubscription implements Serializable, Cloneable, IModelJso
}
}
@Override
public String toString() {
return new ToStringBuilder(this)
.append("myIdElement", myIdElement)
.append("myStatus", myStatus)
.append("myCriteriaString", myCriteriaString)
.append("myEndpointUrl", myEndpointUrl)
.append("myPayloadString", myPayloadString)
// .append("myHeaders", myHeaders)
.append("myChannelType", myChannelType)
// .append("myTrigger", myTrigger)
// .append("myEmailDetails", myEmailDetails)
// .append("myRestHookDetails", myRestHookDetails)
// .append("myChannelExtensions", myChannelExtensions)
.toString();
}
}

View File

@ -25,6 +25,7 @@ import ca.uhn.fhir.interceptor.api.Hook;
import ca.uhn.fhir.interceptor.api.Interceptor;
import ca.uhn.fhir.interceptor.api.Pointcut;
import ca.uhn.fhir.jpa.api.dao.DaoRegistry;
import ca.uhn.fhir.jpa.model.util.JpaConstants;
import ca.uhn.fhir.jpa.subscription.match.matcher.matching.SubscriptionMatchingStrategy;
import ca.uhn.fhir.jpa.subscription.match.matcher.matching.SubscriptionStrategyEvaluator;
import ca.uhn.fhir.jpa.subscription.match.registry.SubscriptionCanonicalizer;
@ -96,32 +97,19 @@ public class SubscriptionValidatingInterceptor {
if (!finished) {
String query = subscription.getCriteriaString();
if (isBlank(query)) {
throw new UnprocessableEntityException("Subscription.criteria must be populated");
}
validateQuery(subscription.getCriteriaString(), "Subscription.criteria");
int sep = query.indexOf('?');
if (sep <= 1) {
throw new UnprocessableEntityException("Subscription.criteria must be in the form \"{Resource Type}?[params]\"");
}
String resType = query.substring(0, sep);
if (resType.contains("/")) {
throw new UnprocessableEntityException("Subscription.criteria must be in the form \"{Resource Type}?[params]\"");
if (subscription.getPayloadSearchResult() != null) {
validateQuery(subscription.getPayloadSearchResult(), "Subscription.extension(url='" + JpaConstants.EXT_SUBSCRIPTION_PAYLOAD_SEARCH_RESULT + "')");
}
validateChannelType(subscription);
if (!myDaoRegistry.isResourceTypeSupported(resType)) {
throw new UnprocessableEntityException("Subscription.criteria contains invalid/unsupported resource type: " + resType);
}
try {
SubscriptionMatchingStrategy strategy = mySubscriptionStrategyEvaluator.determineStrategy(query);
SubscriptionMatchingStrategy strategy = mySubscriptionStrategyEvaluator.determineStrategy(subscription.getCriteriaString());
mySubscriptionCanonicalizer.setMatchingStrategyTag(theSubscription, strategy);
} catch (InvalidRequestException | DataFormatException e) {
throw new UnprocessableEntityException("Invalid subscription criteria submitted: " + query + " " + e.getMessage());
throw new UnprocessableEntityException("Invalid subscription criteria submitted: " + subscription.getCriteriaString() + " " + e.getMessage());
}
if (subscription.getChannelType() == null) {
@ -129,6 +117,28 @@ public class SubscriptionValidatingInterceptor {
} else if (subscription.getChannelType() == CanonicalSubscriptionChannelType.MESSAGE) {
validateMessageSubscriptionEndpoint(subscription.getEndpointUrl());
}
}
}
public void validateQuery(String theQuery, String theFieldName) {
if (isBlank(theQuery)) {
throw new UnprocessableEntityException(theFieldName + " must be populated");
}
int sep = theQuery.indexOf('?');
if (sep <= 1) {
throw new UnprocessableEntityException(theFieldName + " must be in the form \"{Resource Type}?[params]\"");
}
String resType = theQuery.substring(0, sep);
if (resType.contains("/")) {
throw new UnprocessableEntityException(theFieldName + " must be in the form \"{Resource Type}?[params]\"");
}
if (!myDaoRegistry.isResourceTypeSupported(resType)) {
throw new UnprocessableEntityException(theFieldName + " contains invalid/unsupported resource type: " + resType);
}
}

View File

@ -0,0 +1,40 @@
package ca.uhn.fhir.util;
import ca.uhn.fhir.context.FhirContext;
import org.hl7.fhir.r4.model.Bundle;
import org.hl7.fhir.r4.model.Patient;
import org.hl7.fhir.r4.model.codesystems.HttpVerb;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
public class TransactionBuilderTest {
private static final Logger ourLog = LoggerFactory.getLogger(TransactionBuilderTest.class);
private FhirContext myFhirContext = FhirContext.forR4();
@Test
public void testAddEntryUpdate() {
TransactionBuilder builder = new TransactionBuilder(myFhirContext);
Patient patient = new Patient();
patient.setId("http://foo/Patient/123");
patient.setActive(true);
builder.addUpdateEntry(patient);
Bundle bundle = (Bundle) builder.getBundle();
ourLog.info("Bundle:\n{}", myFhirContext.newJsonParser().setPrettyPrint(true).encodeResourceToString(bundle));
assertEquals(Bundle.BundleType.TRANSACTION, bundle.getType());
assertEquals(1, bundle.getEntry().size());
assertSame(patient, bundle.getEntry().get(0).getResource());
assertEquals("http://foo/Patient/123", bundle.getEntry().get(0).getFullUrl());
assertEquals("Patient/123", bundle.getEntry().get(0).getRequest().getUrl());
assertEquals(Bundle.HTTPVerb.PUT, bundle.getEntry().get(0).getRequest().getMethod());
}
}