Merge branch 'master' of github.com:jamesagnew/hapi-fhir
This commit is contained in:
commit
b9b85f5ba9
|
@ -9,9 +9,9 @@ package ca.uhn.fhir.jpa.subscription;
|
|||
* 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.
|
||||
|
@ -32,7 +32,6 @@ import org.springframework.messaging.MessagingException;
|
|||
|
||||
public abstract class BaseSubscriptionDeliverySubscriber extends BaseSubscriptionSubscriber {
|
||||
private static final Logger ourLog = LoggerFactory.getLogger(BaseSubscriptionDeliverySubscriber.class);
|
||||
private boolean myReloadResourceBeforeDelivery = true;
|
||||
|
||||
public BaseSubscriptionDeliverySubscriber(IFhirResourceDao<?> theSubscriptionDao, Subscription.SubscriptionChannelType theChannelType, BaseSubscriptionInterceptor theSubscriptionInterceptor) {
|
||||
super(theSubscriptionDao, theChannelType, theSubscriptionInterceptor);
|
||||
|
@ -51,35 +50,30 @@ public abstract class BaseSubscriptionDeliverySubscriber extends BaseSubscriptio
|
|||
ResourceDeliveryMessage msg = (ResourceDeliveryMessage) theMessage.getPayload();
|
||||
subscriptionId = msg.getSubscription().getIdElement(getContext()).getValue();
|
||||
|
||||
if (!subscriptionTypeApplies(getContext(), msg.getSubscription().getBackingSubscription(getContext()))) {
|
||||
return;
|
||||
}
|
||||
|
||||
CanonicalSubscription updatedSubscription = (CanonicalSubscription) getSubscriptionInterceptor().getIdToSubscription().get(msg.getSubscription().getIdElement(getContext()).getIdPart());
|
||||
if (updatedSubscription != null) {
|
||||
msg.setSubscription(updatedSubscription);
|
||||
}
|
||||
|
||||
if (myReloadResourceBeforeDelivery) {
|
||||
// Reload the payload just in case any interceptors modified
|
||||
// it before it was saved to the database. This is also
|
||||
// useful for resources created in a transaction, since they
|
||||
// can have placeholder IDs in them.
|
||||
IIdType payloadId = msg.getPayloadId(getContext());
|
||||
Class type = getContext().getResourceDefinition(payloadId.getResourceType()).getImplementingClass();
|
||||
IFhirResourceDao dao = getSubscriptionInterceptor().getDao(type);
|
||||
IBaseResource loadedPayload;
|
||||
try {
|
||||
loadedPayload = dao.read(payloadId);
|
||||
} catch (ResourceNotFoundException e) {
|
||||
// This can happen if a last minute failure happens when saving a resource,
|
||||
// eg a constraint causes the transaction to roll back on commit
|
||||
ourLog.warn("Unable to find resource {} - Aborting delivery", payloadId.getValue());
|
||||
return;
|
||||
}
|
||||
msg.setPayload(getContext(), loadedPayload);
|
||||
if (!subscriptionTypeApplies(msg.getSubscription())) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Load the resource
|
||||
IIdType payloadId = msg.getPayloadId(getContext());
|
||||
Class type = getContext().getResourceDefinition(payloadId.getResourceType()).getImplementingClass();
|
||||
IFhirResourceDao dao = getSubscriptionInterceptor().getDao(type);
|
||||
IBaseResource loadedPayload;
|
||||
try {
|
||||
loadedPayload = dao.read(payloadId);
|
||||
} catch (ResourceNotFoundException e) {
|
||||
// This can happen if a last minute failure happens when saving a resource,
|
||||
// eg a constraint causes the transaction to roll back on commit
|
||||
ourLog.warn("Unable to find resource {} - Aborting delivery", payloadId.getValue());
|
||||
return;
|
||||
}
|
||||
msg.setPayload(getContext(), loadedPayload);
|
||||
|
||||
handleMessage(msg);
|
||||
} catch (Exception e) {
|
||||
String msg = "Failure handling subscription payload for subscription: " + subscriptionId;
|
||||
|
@ -90,8 +84,4 @@ public abstract class BaseSubscriptionDeliverySubscriber extends BaseSubscriptio
|
|||
|
||||
public abstract void handleMessage(ResourceDeliveryMessage theMessage) throws Exception;
|
||||
|
||||
public void setReloadResourceBeforeDelivery(boolean theReloadResourceBeforeDelivery) {
|
||||
myReloadResourceBeforeDelivery = theReloadResourceBeforeDelivery;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -70,10 +70,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
|
||||
static final String SUBSCRIPTION_STATUS = "Subscription.status";
|
||||
static final String SUBSCRIPTION_TYPE = "Subscription.channel.type";
|
||||
static final String SUBSCRIPTION_CRITERIA = "Subscription.criteria";
|
||||
static final String SUBSCRIPTION_ENDPOINT = "Subscription.channel.endpoint";
|
||||
static final String SUBSCRIPTION_PAYLOAD = "Subscription.channel.payload";
|
||||
static final String SUBSCRIPTION_HEADER = "Subscription.channel.header";
|
||||
private static final Integer MAX_SUBSCRIPTION_RESULTS = 1000;
|
||||
private SubscribableChannel myProcessingChannel;
|
||||
private SubscribableChannel myDeliveryChannel;
|
||||
|
@ -128,7 +124,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
CanonicalSubscription retVal = new CanonicalSubscription();
|
||||
try {
|
||||
retVal.setStatus(org.hl7.fhir.r4.model.Subscription.SubscriptionStatus.fromCode(subscription.getStatus()));
|
||||
retVal.setBackingSubscription(myCtx, theSubscription);
|
||||
retVal.setChannelType(org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType.fromCode(subscription.getChannel().getType()));
|
||||
retVal.setCriteriaString(subscription.getCriteria());
|
||||
retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
|
||||
|
@ -147,7 +142,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
CanonicalSubscription retVal = new CanonicalSubscription();
|
||||
try {
|
||||
retVal.setStatus(org.hl7.fhir.r4.model.Subscription.SubscriptionStatus.fromCode(subscription.getStatus().toCode()));
|
||||
retVal.setBackingSubscription(myCtx, theSubscription);
|
||||
retVal.setChannelType(org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType.fromCode(subscription.getChannel().getType().toCode()));
|
||||
retVal.setCriteriaString(subscription.getCriteria());
|
||||
retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
|
||||
|
@ -193,7 +187,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
|
||||
CanonicalSubscription retVal = new CanonicalSubscription();
|
||||
retVal.setStatus(subscription.getStatus());
|
||||
retVal.setBackingSubscription(myCtx, theSubscription);
|
||||
retVal.setChannelType(subscription.getChannel().getType());
|
||||
retVal.setCriteriaString(subscription.getCriteria());
|
||||
retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
|
||||
|
@ -204,7 +197,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
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);
|
||||
|
@ -260,8 +252,7 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
|
|||
myResourceTypeToDao = theResourceTypeToDao;
|
||||
}
|
||||
|
||||
IFhirResourceDao<R> dao = (IFhirResourceDao<R>) myResourceTypeToDao.get(theType);
|
||||
return dao;
|
||||
return (IFhirResourceDao<R>) myResourceTypeToDao.get(theType);
|
||||
}
|
||||
|
||||
public SubscribableChannel getDeliveryChannel() {
|
||||
|
|
|
@ -22,8 +22,6 @@ package ca.uhn.fhir.jpa.subscription;
|
|||
|
||||
import ca.uhn.fhir.context.FhirContext;
|
||||
import ca.uhn.fhir.jpa.dao.IFhirResourceDao;
|
||||
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;
|
||||
|
||||
|
@ -62,19 +60,19 @@ public abstract class BaseSubscriptionSubscriber implements MessageHandler {
|
|||
/**
|
||||
* Does this subscription type (e.g. rest hook, websocket, etc) apply to this interceptor?
|
||||
*/
|
||||
protected boolean subscriptionTypeApplies(FhirContext theCtx, IBaseResource theSubscription) {
|
||||
protected boolean subscriptionTypeApplies(CanonicalSubscription theSubscription) {
|
||||
Subscription.SubscriptionChannelType channelType = getChannelType();
|
||||
return subscriptionTypeApplies(theCtx, theSubscription, channelType);
|
||||
String subscriptionType = theSubscription.getChannelType().toCode();
|
||||
return subscriptionTypeApplies(subscriptionType, channelType);
|
||||
}
|
||||
|
||||
/**
|
||||
* Does this subscription type (e.g. rest hook, websocket, etc) apply to this interceptor?
|
||||
*/
|
||||
static boolean subscriptionTypeApplies(FhirContext theCtx, IBaseResource theSubscription, Subscription.SubscriptionChannelType theChannelType) {
|
||||
IPrimitiveType<?> subscriptionType = theCtx.newTerser().getSingleValueOrNull(theSubscription, BaseSubscriptionInterceptor.SUBSCRIPTION_TYPE, IPrimitiveType.class);
|
||||
static boolean subscriptionTypeApplies(String theSubscriptionChannelTypeCode, Subscription.SubscriptionChannelType theChannelType) {
|
||||
boolean subscriptionTypeApplies = false;
|
||||
if (subscriptionType != null) {
|
||||
if (theChannelType.toCode().equals(subscriptionType.getValueAsString())) {
|
||||
if (theSubscriptionChannelTypeCode != null) {
|
||||
if (theChannelType.toCode().equals(theSubscriptionChannelTypeCode)) {
|
||||
subscriptionTypeApplies = true;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -22,12 +22,10 @@ package ca.uhn.fhir.jpa.subscription;
|
|||
|
||||
import ca.uhn.fhir.context.FhirContext;
|
||||
import com.fasterxml.jackson.annotation.JsonAutoDetect;
|
||||
import com.fasterxml.jackson.annotation.JsonIgnore;
|
||||
import com.fasterxml.jackson.annotation.JsonInclude;
|
||||
import com.fasterxml.jackson.annotation.JsonProperty;
|
||||
import org.apache.commons.lang3.builder.EqualsBuilder;
|
||||
import org.apache.commons.lang3.builder.HashCodeBuilder;
|
||||
import org.hl7.fhir.instance.model.api.IBaseResource;
|
||||
import org.hl7.fhir.instance.model.api.IIdType;
|
||||
import org.hl7.fhir.instance.model.api.IPrimitiveType;
|
||||
import org.hl7.fhir.r4.model.EventDefinition;
|
||||
|
@ -59,10 +57,6 @@ public class CanonicalSubscription implements Serializable {
|
|||
private Subscription.SubscriptionChannelType myChannelType;
|
||||
@JsonProperty("status")
|
||||
private Subscription.SubscriptionStatus myStatus;
|
||||
@JsonIgnore
|
||||
private transient IBaseResource myBackingSubscription;
|
||||
@JsonProperty("backingSubscription")
|
||||
private String myBackingSubscriptionString;
|
||||
@JsonProperty("triggerDefinition")
|
||||
private CanonicalEventDefinition myTrigger;
|
||||
@JsonProperty("emailDetails")
|
||||
|
@ -91,13 +85,6 @@ public class CanonicalSubscription implements Serializable {
|
|||
.isEquals();
|
||||
}
|
||||
|
||||
public IBaseResource getBackingSubscription(FhirContext theCtx) {
|
||||
if (myBackingSubscription == null && myBackingSubscriptionString != null) {
|
||||
myBackingSubscription = theCtx.newJsonParser().parseResource(myBackingSubscriptionString);
|
||||
}
|
||||
return myBackingSubscription;
|
||||
}
|
||||
|
||||
public Subscription.SubscriptionChannelType getChannelType() {
|
||||
return myChannelType;
|
||||
}
|
||||
|
@ -190,14 +177,6 @@ public class CanonicalSubscription implements Serializable {
|
|||
.toHashCode();
|
||||
}
|
||||
|
||||
public void setBackingSubscription(FhirContext theCtx, IBaseResource theBackingSubscription) {
|
||||
myBackingSubscription = theBackingSubscription;
|
||||
myBackingSubscriptionString = null;
|
||||
if (myBackingSubscription != null) {
|
||||
myBackingSubscriptionString = theCtx.newJsonParser().encodeResourceToString(myBackingSubscription);
|
||||
}
|
||||
}
|
||||
|
||||
public void setHeaders(List<? extends IPrimitiveType<String>> theHeader) {
|
||||
myHeaders = new ArrayList<>();
|
||||
for (IPrimitiveType<String> next : theHeader) {
|
||||
|
|
|
@ -24,6 +24,7 @@ import ca.uhn.fhir.context.FhirContext;
|
|||
import ca.uhn.fhir.rest.api.RestOperationTypeEnum;
|
||||
import com.fasterxml.jackson.annotation.*;
|
||||
import com.google.gson.Gson;
|
||||
import org.apache.commons.lang3.Validate;
|
||||
import org.hl7.fhir.instance.model.api.IBaseResource;
|
||||
import org.hl7.fhir.instance.model.api.IIdType;
|
||||
|
||||
|
@ -41,8 +42,6 @@ public class ResourceDeliveryMessage {
|
|||
private String mySubscriptionString;
|
||||
@JsonIgnore
|
||||
private transient IBaseResource myPayload;
|
||||
@JsonProperty("payload")
|
||||
private String myPayoadString;
|
||||
@JsonProperty("payloadId")
|
||||
private String myPayloadId;
|
||||
@JsonProperty("operationType")
|
||||
|
@ -57,9 +56,7 @@ public class ResourceDeliveryMessage {
|
|||
}
|
||||
|
||||
public IBaseResource getPayload(FhirContext theCtx) {
|
||||
if (myPayload == null && myPayoadString != null) {
|
||||
myPayload = theCtx.newJsonParser().parseResource(myPayoadString);
|
||||
}
|
||||
Validate.notNull(myPayload);
|
||||
return myPayload;
|
||||
}
|
||||
|
||||
|
@ -87,7 +84,6 @@ public class ResourceDeliveryMessage {
|
|||
|
||||
public void setPayload(FhirContext theCtx, IBaseResource thePayload) {
|
||||
myPayload = thePayload;
|
||||
myPayoadString = theCtx.newJsonParser().encodeResourceToString(thePayload);
|
||||
}
|
||||
|
||||
public void setPayloadId(IIdType thePayloadId) {
|
||||
|
|
|
@ -35,7 +35,6 @@ import org.slf4j.Logger;
|
|||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.core.task.AsyncTaskExecutor;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.TransactionStatus;
|
||||
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
|
||||
|
@ -43,7 +42,6 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter
|
|||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
|
@ -73,7 +71,14 @@ public class SubscriptionActivatingSubscriber {
|
|||
}
|
||||
|
||||
public void activateAndRegisterSubscriptionIfRequired(final IBaseResource theSubscription) {
|
||||
boolean subscriptionTypeApplies = BaseSubscriptionSubscriber.subscriptionTypeApplies(myCtx, theSubscription, myChannelType);
|
||||
|
||||
// Grab the value for "Subscription.channel.type" so we can see if this
|
||||
// subscriber applies..
|
||||
String subscriptionChannelType = myCtx
|
||||
.newTerser()
|
||||
.getSingleValueOrNull(theSubscription, BaseSubscriptionInterceptor.SUBSCRIPTION_TYPE, IPrimitiveType.class)
|
||||
.getValueAsString();
|
||||
boolean subscriptionTypeApplies = BaseSubscriptionSubscriber.subscriptionTypeApplies(subscriptionChannelType, myChannelType);
|
||||
if (subscriptionTypeApplies == false) {
|
||||
return;
|
||||
}
|
||||
|
|
2
pom.xml
2
pom.xml
|
@ -1431,7 +1431,7 @@
|
|||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>versions-maven-plugin</artifactId>
|
||||
<version>2.5</version>
|
||||
<version>2.6-SNAPSHOT</version>
|
||||
<configuration>
|
||||
<processDependencyManagementTransitive>false</processDependencyManagementTransitive>
|
||||
</configuration>
|
||||
|
|
|
@ -244,6 +244,11 @@
|
|||
a resource could be associated with the wrong entry in the response.
|
||||
Thanks to GitHub user @jbalbien for the pull request!
|
||||
</action>
|
||||
<action type="add">
|
||||
JPA subscription delivery queues no longer store the resource body in the
|
||||
queue (only the ID), which should reduce the memory/disk footprint of the queue
|
||||
when it grows long.
|
||||
</action>
|
||||
</release>
|
||||
<release version="3.4.0" date="2018-05-28">
|
||||
<action type="add">
|
||||
|
|
Loading…
Reference in New Issue