
001package ca.uhn.fhir.jpa.subscription.match.matcher.subscriber; 002 003/*- 004 * #%L 005 * HAPI FHIR Subscription Server 006 * %% 007 * Copyright (C) 2014 - 2022 Smile CDR, Inc. 008 * %% 009 * Licensed under the Apache License, Version 2.0 (the "License"); 010 * you may not use this file except in compliance with the License. 011 * You may obtain a copy of the License at 012 * 013 * http://www.apache.org/licenses/LICENSE-2.0 014 * 015 * Unless required by applicable law or agreed to in writing, software 016 * distributed under the License is distributed on an "AS IS" BASIS, 017 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 018 * See the License for the specific language governing permissions and 019 * limitations under the License. 020 * #L% 021 */ 022 023import ca.uhn.fhir.context.FhirContext; 024import ca.uhn.fhir.interceptor.model.RequestPartitionId; 025import ca.uhn.fhir.jpa.api.dao.DaoRegistry; 026import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao; 027import ca.uhn.fhir.jpa.partition.SystemRequestDetails; 028import ca.uhn.fhir.jpa.subscription.match.registry.SubscriptionCanonicalizer; 029import ca.uhn.fhir.jpa.subscription.match.registry.SubscriptionRegistry; 030import ca.uhn.fhir.jpa.subscription.model.ResourceModifiedJsonMessage; 031import ca.uhn.fhir.jpa.subscription.model.ResourceModifiedMessage; 032import ca.uhn.fhir.rest.api.server.RequestDetails; 033import ca.uhn.fhir.rest.server.exceptions.ResourceGoneException; 034import org.hl7.fhir.instance.model.api.IBaseResource; 035import org.hl7.fhir.instance.model.api.IIdType; 036import org.slf4j.Logger; 037import org.slf4j.LoggerFactory; 038import org.springframework.beans.factory.annotation.Autowired; 039import org.springframework.messaging.Message; 040import org.springframework.messaging.MessageHandler; 041import org.springframework.messaging.MessagingException; 042 043import javax.annotation.Nonnull; 044 045/** 046 * Responsible for transitioning subscription resources from REQUESTED to ACTIVE 047 * Once activated, the subscription is added to the SubscriptionRegistry. 048 * <p> 049 * Also validates criteria. If invalid, rejects the subscription without persisting the subscription. 050 */ 051public class SubscriptionRegisteringSubscriber extends BaseSubscriberForSubscriptionResources implements MessageHandler { 052 private static final Logger ourLog = LoggerFactory.getLogger(SubscriptionRegisteringSubscriber.class); 053 @Autowired 054 private FhirContext myFhirContext; 055 @Autowired 056 private SubscriptionRegistry mySubscriptionRegistry; 057 @Autowired 058 private SubscriptionCanonicalizer mySubscriptionCanonicalizer; 059 @Autowired 060 private DaoRegistry myDaoRegistry; 061 062 /** 063 * Constructor 064 */ 065 public SubscriptionRegisteringSubscriber() { 066 super(); 067 } 068 069 @Override 070 public void handleMessage(@Nonnull Message<?> theMessage) throws MessagingException { 071 if (!(theMessage instanceof ResourceModifiedJsonMessage)) { 072 ourLog.warn("Received message of unexpected type on matching channel: {}", theMessage); 073 return; 074 } 075 076 ResourceModifiedMessage payload = ((ResourceModifiedJsonMessage) theMessage).getPayload(); 077 078 if (!isSubscription(payload)) { 079 return; 080 } 081 082 switch (payload.getOperationType()) { 083 case MANUALLY_TRIGGERED: 084 case TRANSACTION: 085 return; 086 case CREATE: 087 case UPDATE: 088 case DELETE: 089 break; 090 } 091 092 // We read the resource back from the DB instead of using the supplied copy for 093 // two reasons: 094 // - in order to store partition id in the userdata of the resource for partitioned subscriptions 095 // - in case we're processing out of order and a create-then-delete has been processed backwards (or vice versa) 096 097 IBaseResource payloadResource; 098 IIdType payloadId = payload.getPayloadId(myFhirContext).toUnqualifiedVersionless(); 099 try { 100 IFhirResourceDao<?> subscriptionDao = myDaoRegistry.getResourceDao("Subscription"); 101 RequestDetails systemRequestDetails = getPartitionAwareRequestDetails(payload); 102 payloadResource = subscriptionDao.read(payloadId, systemRequestDetails); 103 if (payloadResource == null) { 104 // Only for unit test 105 payloadResource = payload.getPayload(myFhirContext); 106 } 107 } catch (ResourceGoneException e) { 108 mySubscriptionRegistry.unregisterSubscriptionIfRegistered(payloadId.getIdPart()); 109 return; 110 } 111 112 String statusString = mySubscriptionCanonicalizer.getSubscriptionStatus(payloadResource); 113 if ("active".equals(statusString)) { 114 mySubscriptionRegistry.registerSubscriptionUnlessAlreadyRegistered(payloadResource); 115 } else { 116 mySubscriptionRegistry.unregisterSubscriptionIfRegistered(payloadId.getIdPart()); 117 } 118 119 } 120 121 /** 122 * There were some situations where the RequestDetails attempted to use the default partition 123 * and the partition name was a list containing null values (i.e. using the package installer to STORE_AND_INSTALL 124 * Subscriptions while partitioning was enabled). If any partition matches these criteria, 125 * {@link RequestPartitionId#defaultPartition()} is used to obtain the default partition. 126 */ 127 private RequestDetails getPartitionAwareRequestDetails(ResourceModifiedMessage payload) { 128 RequestPartitionId payloadPartitionId = payload.getPartitionId(); 129 if (payloadPartitionId == null || payloadPartitionId.isDefaultPartition()) { 130 // This may look redundant but the package installer STORE_AND_INSTALL Subscriptions when partitioning is enabled 131 // creates a corrupt default partition. This resets it to a clean one. 132 payloadPartitionId = RequestPartitionId.defaultPartition(); 133 } 134 return new SystemRequestDetails().setRequestPartitionId(payloadPartitionId); 135 } 136 137}