honeymoose
/
hapi-fhir
mirror of https://github.com/hapifhir/hapi-fhir.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Work on subscriptions

This commit is contained in:
jamesagnew 2018-08-03 11:56:06 -04:00
parent d07fd76d60
commit 0556dfa7af
5 changed files with 19 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -27,7 +27,6 @@ import org.hl7.fhir.instance.model.api.IIdType;
import org.hl7.fhir.r4.model.Subscription; import org.hl7.fhir.r4.model.Subscription;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.lang.NonNullApi;
import org.springframework.messaging.Message; import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException; import org.springframework.messaging.MessagingException;
@ -51,15 +50,15 @@ public abstract class BaseSubscriptionDeliverySubscriber extends BaseSubscriptio
ResourceDeliveryMessage msg = (ResourceDeliveryMessage) theMessage.getPayload(); ResourceDeliveryMessage msg = (ResourceDeliveryMessage) theMessage.getPayload();
subscriptionId = msg.getSubscription().getIdElement(getContext()).getValue(); 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()); CanonicalSubscription updatedSubscription = (CanonicalSubscription) getSubscriptionInterceptor().getIdToSubscription().get(msg.getSubscription().getIdElement(getContext()).getIdPart());
if (updatedSubscription != null) { if (updatedSubscription != null) {
msg.setSubscription(updatedSubscription); msg.setSubscription(updatedSubscription);
} }
if (!subscriptionTypeApplies(msg.getSubscription())) {
return;
}
// Load the resource // Load the resource
IIdType payloadId = msg.getPayloadId(getContext()); IIdType payloadId = msg.getPayloadId(getContext());
Class type = getContext().getResourceDefinition(payloadId.getResourceType()).getImplementingClass(); Class type = getContext().getResourceDefinition(payloadId.getResourceType()).getImplementingClass();

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

@ -70,10 +70,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
static final String SUBSCRIPTION_STATUS = "Subscription.status"; static final String SUBSCRIPTION_STATUS = "Subscription.status";
static final String SUBSCRIPTION_TYPE = "Subscription.channel.type"; 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 static final Integer MAX_SUBSCRIPTION_RESULTS = 1000;
private SubscribableChannel myProcessingChannel; private SubscribableChannel myProcessingChannel;
private SubscribableChannel myDeliveryChannel; private SubscribableChannel myDeliveryChannel;
@ -128,7 +124,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
CanonicalSubscription retVal = new CanonicalSubscription(); CanonicalSubscription retVal = new CanonicalSubscription();
try { try {
retVal.setStatus(org.hl7.fhir.r4.model.Subscription.SubscriptionStatus.fromCode(subscription.getStatus())); 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.setChannelType(org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType.fromCode(subscription.getChannel().getType()));
retVal.setCriteriaString(subscription.getCriteria()); retVal.setCriteriaString(subscription.getCriteria());
retVal.setEndpointUrl(subscription.getChannel().getEndpoint()); retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
@ -147,7 +142,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
CanonicalSubscription retVal = new CanonicalSubscription(); CanonicalSubscription retVal = new CanonicalSubscription();
try { try {
retVal.setStatus(org.hl7.fhir.r4.model.Subscription.SubscriptionStatus.fromCode(subscription.getStatus().toCode())); 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.setChannelType(org.hl7.fhir.r4.model.Subscription.SubscriptionChannelType.fromCode(subscription.getChannel().getType().toCode()));
retVal.setCriteriaString(subscription.getCriteria()); retVal.setCriteriaString(subscription.getCriteria());
retVal.setEndpointUrl(subscription.getChannel().getEndpoint()); retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
@ -193,7 +187,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
CanonicalSubscription retVal = new CanonicalSubscription(); CanonicalSubscription retVal = new CanonicalSubscription();
retVal.setStatus(subscription.getStatus()); retVal.setStatus(subscription.getStatus());
retVal.setBackingSubscription(myCtx, theSubscription);
retVal.setChannelType(subscription.getChannel().getType()); retVal.setChannelType(subscription.getChannel().getType());
retVal.setCriteriaString(subscription.getCriteria()); retVal.setCriteriaString(subscription.getCriteria());
retVal.setEndpointUrl(subscription.getChannel().getEndpoint()); retVal.setEndpointUrl(subscription.getChannel().getEndpoint());
@ -204,7 +197,6 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
if (retVal.getChannelType() == Subscription.SubscriptionChannelType.EMAIL) { if (retVal.getChannelType() == Subscription.SubscriptionChannelType.EMAIL) {
String from; String from;
String subjectTemplate; String subjectTemplate;
String bodyTemplate;
try { try {
from = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_EMAIL_FROM); from = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_EMAIL_FROM);
subjectTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_SUBJECT_TEMPLATE); subjectTemplate = subscription.getChannel().getExtensionString(JpaConstants.EXT_SUBSCRIPTION_SUBJECT_TEMPLATE);
@ -260,8 +252,7 @@ public abstract class BaseSubscriptionInterceptor<S extends IBaseResource> exten
myResourceTypeToDao = theResourceTypeToDao; myResourceTypeToDao = theResourceTypeToDao;
} }
IFhirResourceDao<R> dao = (IFhirResourceDao<R>) myResourceTypeToDao.get(theType); return (IFhirResourceDao<R>) myResourceTypeToDao.get(theType);
return dao;
} }
public SubscribableChannel getDeliveryChannel() { public SubscribableChannel getDeliveryChannel() {

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

@ -22,8 +22,6 @@ package ca.uhn.fhir.jpa.subscription;
import ca.uhn.fhir.context.FhirContext; import ca.uhn.fhir.context.FhirContext;
import ca.uhn.fhir.jpa.dao.IFhirResourceDao; 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.hl7.fhir.r4.model.Subscription;
import org.springframework.messaging.MessageHandler; 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? * 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(); 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? * Does this subscription type (e.g. rest hook, websocket, etc) apply to this interceptor?
*/ */
static boolean subscriptionTypeApplies(FhirContext theCtx, IBaseResource theSubscription, Subscription.SubscriptionChannelType theChannelType) { static boolean subscriptionTypeApplies(String theSubscriptionChannelTypeCode, Subscription.SubscriptionChannelType theChannelType) {
IPrimitiveType<?> subscriptionType = theCtx.newTerser().getSingleValueOrNull(theSubscription, BaseSubscriptionInterceptor.SUBSCRIPTION_TYPE, IPrimitiveType.class);
boolean subscriptionTypeApplies = false; boolean subscriptionTypeApplies = false;
if (subscriptionType != null) { if (theSubscriptionChannelTypeCode != null) {
if (theChannelType.toCode().equals(subscriptionType.getValueAsString())) { if (theChannelType.toCode().equals(theSubscriptionChannelTypeCode)) {
subscriptionTypeApplies = true; subscriptionTypeApplies = true;
} }
} }

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

@ -22,12 +22,10 @@ package ca.uhn.fhir.jpa.subscription;
import ca.uhn.fhir.context.FhirContext; import ca.uhn.fhir.context.FhirContext;
import com.fasterxml.jackson.annotation.JsonAutoDetect; import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.commons.lang3.builder.EqualsBuilder; import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.commons.lang3.builder.HashCodeBuilder; 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.IIdType;
import org.hl7.fhir.instance.model.api.IPrimitiveType; import org.hl7.fhir.instance.model.api.IPrimitiveType;
import org.hl7.fhir.r4.model.EventDefinition; import org.hl7.fhir.r4.model.EventDefinition;
@ -59,10 +57,6 @@ public class CanonicalSubscription implements Serializable {
private Subscription.SubscriptionChannelType myChannelType; private Subscription.SubscriptionChannelType myChannelType;
@JsonProperty("status") @JsonProperty("status")
private Subscription.SubscriptionStatus myStatus; private Subscription.SubscriptionStatus myStatus;
@JsonIgnore
private transient IBaseResource myBackingSubscription;
@JsonProperty("backingSubscription")
private String myBackingSubscriptionString;
@JsonProperty("triggerDefinition") @JsonProperty("triggerDefinition")
private CanonicalEventDefinition myTrigger; private CanonicalEventDefinition myTrigger;
@JsonProperty("emailDetails") @JsonProperty("emailDetails")
@ -91,13 +85,6 @@ public class CanonicalSubscription implements Serializable {
.isEquals(); .isEquals();
} }
public IBaseResource getBackingSubscription(FhirContext theCtx) {
if (myBackingSubscription == null && myBackingSubscriptionString != null) {
myBackingSubscription = theCtx.newJsonParser().parseResource(myBackingSubscriptionString);
}
return myBackingSubscription;
}
public Subscription.SubscriptionChannelType getChannelType() { public Subscription.SubscriptionChannelType getChannelType() {
return myChannelType; return myChannelType;
} }
@ -190,14 +177,6 @@ public class CanonicalSubscription implements Serializable {
.toHashCode(); .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) { public void setHeaders(List<? extends IPrimitiveType<String>> theHeader) {
myHeaders = new ArrayList<>(); myHeaders = new ArrayList<>();
for (IPrimitiveType<String> next : theHeader) { for (IPrimitiveType<String> next : theHeader) {

11
hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/subscription/SubscriptionActivatingSubscriber.java
View File

@ -35,7 +35,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.core.task.AsyncTaskExecutor; import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.messaging.MessagingException; import org.springframework.messaging.MessagingException;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallbackWithoutResult; 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.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate; import org.springframework.transaction.support.TransactionTemplate;
import java.util.Date;
import java.util.concurrent.Future; import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -73,7 +71,14 @@ public class SubscriptionActivatingSubscriber {
} }
public void activateAndRegisterSubscriptionIfRequired(final IBaseResource theSubscription) { 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) { if (subscriptionTypeApplies == false) {
return; return;
} }