refactor

hapi-fhir-jpaserver-subscription/src/main/java/ca/uhn/fhir/jpa/subscription/match/matcher/subscriber/SubscriptionMatchingSubscriber.java

@@ -122,98 +122,107 @@ private void doMatchActiveSubscriptionsAndDeliver(ResourceModifiedMessage theMsg
  Collection<ActiveSubscription> subscriptions = mySubscriptionRegistry.getAll();
 
  ourLog.trace("Testing {} subscriptions for applicability", subscriptions.size());
- boolean resourceMatched = false;
+ boolean anySubscriptionsMatchedResource = false;
 
  for (ActiveSubscription nextActiveSubscription : subscriptions) {
- // skip if the partitions don't match
- CanonicalSubscription subscription = nextActiveSubscription.getSubscription();
- if (subscription != null && theMsg.getPartitionId() != null &&
- theMsg.getPartitionId().hasPartitionIds() && !subscription.getCrossPartitionEnabled() &&
- !theMsg.getPartitionId().hasPartitionId(subscription.getRequestPartitionId())) {
- continue;
- }
- String nextSubscriptionId = getId(nextActiveSubscription);
-
- if (isNotBlank(theMsg.getSubscriptionId())) {
- if (!theMsg.getSubscriptionId().equals(nextSubscriptionId)) {
- // TODO KHS we should use a hash to look it up instead of this full table scan
- ourLog.debug("Ignoring subscription {} because it is not {}", nextSubscriptionId, theMsg.getSubscriptionId());
- continue;
- }
- }
+ anySubscriptionsMatchedResource |= processSubscription(theMsg, resourceId, nextActiveSubscription);
+ }
 
- if (!resourceTypeIsAppropriateForSubscription(nextActiveSubscription, resourceId)) {
- continue;
- }
+ if (!anySubscriptionsMatchedResource) {
+ // Interceptor call: SUBSCRIPTION_RESOURCE_MATCHED
+ HookParams params = new HookParams()
+ .add(ResourceModifiedMessage.class, theMsg);
+ myInterceptorBroadcaster.callHooks(Pointcut.SUBSCRIPTION_RESOURCE_DID_NOT_MATCH_ANY_SUBSCRIPTIONS, params);
+ }
+ }
 
- if (theMsg.getOperationType().equals(DELETE)) {
- if (!nextActiveSubscription.getSubscription().getSendDeleteMessages()) {
- ourLog.trace("Not processing modified message for {}", theMsg.getOperationType());
- continue;
- }
- }
+ /**
+ * Returns true if subscription matched, and processing completed successfully, and the message was sent to the delivery channel. False otherwise.
+ *
+ */
+ private boolean processSubscription(ResourceModifiedMessage theMsg, IIdType theResourceId, ActiveSubscription theActiveSubscription) {
+ // skip if the partitions don't match
+ CanonicalSubscription subscription = theActiveSubscription.getSubscription();
+ if (subscription != null && theMsg.getPartitionId() != null &&
+ theMsg.getPartitionId().hasPartitionIds() && !subscription.getCrossPartitionEnabled() &&
+ !theMsg.getPartitionId().hasPartitionId(subscription.getRequestPartitionId())) {
+ return false;
+ }
+ String nextSubscriptionId = getId(theActiveSubscription);
 
- InMemoryMatchResult matchResult;
- if (nextActiveSubscription.getCriteria().getType() == SubscriptionCriteriaParser.TypeEnum.SEARCH_EXPRESSION) {
- matchResult = mySubscriptionMatcher.match(nextActiveSubscription.getSubscription(), theMsg);
- if (!matchResult.matched()) {
- ourLog.trace("Subscription {} was not matched by resource {} {}",
- nextActiveSubscription.getId(),
- resourceId.toUnqualifiedVersionless().getValue(),
- matchResult.isInMemory() ? "in-memory" : "by querying the repository");
- continue;
- }
- ourLog.debug("Subscription {} was matched by resource {} {}",
- nextActiveSubscription.getId(),
- resourceId.toUnqualifiedVersionless().getValue(),
- matchResult.isInMemory() ? "in-memory" : "by querying the repository");
- } else {
- ourLog.trace("Subscription {} was not matched by resource {} - No search expression",
- nextActiveSubscription.getId(),
- resourceId.toUnqualifiedVersionless().getValue());
- matchResult = InMemoryMatchResult.successfulMatch();
- matchResult.setInMemory(true);
+ if (isNotBlank(theMsg.getSubscriptionId())) {
+ if (!theMsg.getSubscriptionId().equals(nextSubscriptionId)) {
+ // TODO KHS we should use a hash to look it up instead of this full table scan
+ ourLog.debug("Ignoring subscription {} because it is not {}", nextSubscriptionId, theMsg.getSubscriptionId());
+ return false;
  }
+ }
+
+ if (!resourceTypeIsAppropriateForSubscription(theActiveSubscription, theResourceId)) {
+ return false;
+ }
 
- IBaseResource payload = theMsg.getNewPayload(myFhirContext);
+ if (theMsg.getOperationType().equals(DELETE)) {
+ if (!theActiveSubscription.getSubscription().getSendDeleteMessages()) {
+ ourLog.trace("Not processing modified message for {}", theMsg.getOperationType());
+ return false;
+ }
+ }
 
- EncodingEnum encoding = null;
- if (subscription != null && subscription.getPayloadString() != null && !subscription.getPayloadString().isEmpty()) {
- encoding = EncodingEnum.forContentType(subscription.getPayloadString());
+ InMemoryMatchResult matchResult;
+ if (theActiveSubscription.getCriteria().getType() == SubscriptionCriteriaParser.TypeEnum.SEARCH_EXPRESSION) {
+ matchResult = mySubscriptionMatcher.match(theActiveSubscription.getSubscription(), theMsg);
+ if (!matchResult.matched()) {
+ ourLog.trace("Subscription {} was not matched by resource {} {}",
+ theActiveSubscription.getId(),
+ theResourceId.toUnqualifiedVersionless().getValue(),
+ matchResult.isInMemory() ? "in-memory" : "by querying the repository");
+ return false;
  }
- encoding = defaultIfNull(encoding, EncodingEnum.JSON);
+ ourLog.debug("Subscription {} was matched by resource {} {}",
+ theActiveSubscription.getId(),
+ theResourceId.toUnqualifiedVersionless().getValue(),
+ matchResult.isInMemory() ? "in-memory" : "by querying the repository");
+ } else {
+ ourLog.trace("Subscription {} was not matched by resource {} - No search expression",
+ theActiveSubscription.getId(),
+ theResourceId.toUnqualifiedVersionless().getValue());
+ matchResult = InMemoryMatchResult.successfulMatch();
+ matchResult.setInMemory(true);
+ }
 
- ResourceDeliveryMessage deliveryMsg = new ResourceDeliveryMessage();
- deliveryMsg.setPartitionId(theMsg.getPartitionId());
+ IBaseResource payload = theMsg.getNewPayload(myFhirContext);
 
- if (payload != null) {
- deliveryMsg.setPayload(myFhirContext, payload, encoding);
- } else {
- deliveryMsg.setPayloadId(theMsg.getPayloadId(myFhirContext));
- }
- deliveryMsg.setSubscription(subscription);
- deliveryMsg.setOperationType(theMsg.getOperationType());
- deliveryMsg.setTransactionId(theMsg.getTransactionId());
- deliveryMsg.copyAdditionalPropertiesFrom(theMsg);
+ EncodingEnum encoding = null;
+ if (subscription != null && subscription.getPayloadString() != null && !subscription.getPayloadString().isEmpty()) {
+ encoding = EncodingEnum.forContentType(subscription.getPayloadString());
+ }
+ encoding = defaultIfNull(encoding, EncodingEnum.JSON);
 
- // Interceptor call: SUBSCRIPTION_RESOURCE_MATCHED
- HookParams params = new HookParams()
- .add(CanonicalSubscription.class, nextActiveSubscription.getSubscription())
- .add(ResourceDeliveryMessage.class, deliveryMsg)
- .add(InMemoryMatchResult.class, matchResult);
- if (!myInterceptorBroadcaster.callHooks(Pointcut.SUBSCRIPTION_RESOURCE_MATCHED, params)) {
- return;
- }
+ ResourceDeliveryMessage deliveryMsg = new ResourceDeliveryMessage();
+ deliveryMsg.setPartitionId(theMsg.getPartitionId());
 
- resourceMatched |= sendToDeliveryChannel(nextActiveSubscription, deliveryMsg);
+ if (payload != null) {
+ deliveryMsg.setPayload(myFhirContext, payload, encoding);
+ } else {
+ deliveryMsg.setPayloadId(theMsg.getPayloadId(myFhirContext));
  }
+ deliveryMsg.setSubscription(subscription);
+ deliveryMsg.setOperationType(theMsg.getOperationType());
+ deliveryMsg.setTransactionId(theMsg.getTransactionId());
+ deliveryMsg.copyAdditionalPropertiesFrom(theMsg);
 
- if (!resourceMatched) {
- // Interceptor call: SUBSCRIPTION_RESOURCE_MATCHED
- HookParams params = new HookParams()
- .add(ResourceModifiedMessage.class, theMsg);
- myInterceptorBroadcaster.callHooks(Pointcut.SUBSCRIPTION_RESOURCE_DID_NOT_MATCH_ANY_SUBSCRIPTIONS, params);
+ // Interceptor call: SUBSCRIPTION_RESOURCE_MATCHED
+ HookParams params = new HookParams()
+ .add(CanonicalSubscription.class, theActiveSubscription.getSubscription())
+ .add(ResourceDeliveryMessage.class, deliveryMsg)
+ .add(InMemoryMatchResult.class, matchResult);
+ if (!myInterceptorBroadcaster.callHooks(Pointcut.SUBSCRIPTION_RESOURCE_MATCHED, params)) {
+ ourLog.info("Interceptor has decided to abort processing of subscription {}", nextSubscriptionId);
+ return false;
  }
+
+ return sendToDeliveryChannel(theActiveSubscription, deliveryMsg);
  }
 
  private boolean sendToDeliveryChannel(ActiveSubscription nextActiveSubscription, ResourceDeliveryMessage theDeliveryMsg) {