001/*-
002 * #%L
003 * HAPI FHIR Subscription Server
004 * %%
005 * Copyright (C) 2014 - 2024 Smile CDR, Inc.
006 * %%
007 * Licensed under the Apache License, Version 2.0 (the "License");
008 * you may not use this file except in compliance with the License.
009 * You may obtain a copy of the License at
010 *
011 *      http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing, software
014 * distributed under the License is distributed on an "AS IS" BASIS,
015 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
016 * See the License for the specific language governing permissions and
017 * limitations under the License.
018 * #L%
019 */
020package ca.uhn.fhir.jpa.subscription.match.registry;
021
022import ca.uhn.fhir.cache.BaseResourceCacheSynchronizer;
023import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
024import ca.uhn.fhir.jpa.subscription.match.matcher.subscriber.SubscriptionActivatingSubscriber;
025import ca.uhn.fhir.rest.param.TokenOrListParam;
026import ca.uhn.fhir.rest.param.TokenParam;
027import ca.uhn.fhir.rest.server.util.ISearchParamRegistry;
028import ca.uhn.fhir.subscription.SubscriptionConstants;
029import com.google.common.annotations.VisibleForTesting;
030import jakarta.annotation.Nonnull;
031import org.apache.commons.lang3.StringUtils;
032import org.hl7.fhir.instance.model.api.IBaseResource;
033import org.hl7.fhir.r4.model.Subscription;
034import org.slf4j.Logger;
035import org.slf4j.LoggerFactory;
036import org.springframework.beans.factory.annotation.Autowired;
037
038import java.util.HashSet;
039import java.util.List;
040import java.util.Set;
041
042public class SubscriptionLoader extends BaseResourceCacheSynchronizer {
043        private static final Logger ourLog = LoggerFactory.getLogger(SubscriptionLoader.class);
044
045        @Autowired
046        private SubscriptionRegistry mySubscriptionRegistry;
047
048        @Autowired
049        private SubscriptionActivatingSubscriber mySubscriptionActivatingInterceptor;
050
051        @Autowired
052        private SubscriptionCanonicalizer mySubscriptionCanonicalizer;
053
054        @Autowired
055        protected ISearchParamRegistry mySearchParamRegistry;
056
057        /**
058         * Constructor
059         */
060        public SubscriptionLoader() {
061                super("Subscription");
062        }
063
064        @VisibleForTesting
065        public int doSyncSubscriptionsForUnitTest() {
066                return super.doSyncResourcesForUnitTest();
067        }
068
069        @Override
070        @Nonnull
071        protected SearchParameterMap getSearchParameterMap() {
072                SearchParameterMap map = new SearchParameterMap();
073
074                if (mySearchParamRegistry.getActiveSearchParam("Subscription", "status") != null) {
075                        map.add(
076                                        Subscription.SP_STATUS,
077                                        new TokenOrListParam()
078                                                        .addOr(new TokenParam(null, Subscription.SubscriptionStatus.REQUESTED.toCode()))
079                                                        .addOr(new TokenParam(null, Subscription.SubscriptionStatus.ACTIVE.toCode())));
080                }
081                map.setLoadSynchronousUpTo(SubscriptionConstants.MAX_SUBSCRIPTION_RESULTS);
082                return map;
083        }
084
085        @Override
086        protected void handleInit(List<IBaseResource> resourceList) {
087                updateSubscriptionRegistry(resourceList);
088        }
089
090        @Override
091        protected int syncResourcesIntoCache(List<IBaseResource> resourceList) {
092                return updateSubscriptionRegistry(resourceList);
093        }
094
095        private int updateSubscriptionRegistry(List<IBaseResource> theResourceList) {
096                Set<String> allIds = new HashSet<>();
097                int activatedCount = 0;
098                int registeredCount = 0;
099
100                for (IBaseResource resource : theResourceList) {
101                        String nextId = resource.getIdElement().getIdPart();
102                        allIds.add(nextId);
103
104                        boolean activated = activateSubscriptionIfRequested(resource);
105                        if (activated) {
106                                ++activatedCount;
107                        }
108
109                        boolean registered = mySubscriptionRegistry.registerSubscriptionUnlessAlreadyRegistered(resource);
110                        if (registered) {
111                                registeredCount++;
112                        }
113                }
114
115                mySubscriptionRegistry.unregisterAllSubscriptionsNotInCollection(allIds);
116                ourLog.debug(
117                                "Finished sync subscriptions - activated {} and registered {}",
118                                theResourceList.size(),
119                                registeredCount);
120                return activatedCount;
121        }
122
123        /**
124         * Check status of theSubscription and update to "active" if needed.
125         * @return true if activated
126         */
127        private boolean activateSubscriptionIfRequested(IBaseResource theSubscription) {
128                boolean successfullyActivated = false;
129
130                if (SubscriptionConstants.REQUESTED_STATUS.equals(
131                                mySubscriptionCanonicalizer.getSubscriptionStatus(theSubscription))) {
132                        if (mySubscriptionActivatingInterceptor.isChannelTypeSupported(theSubscription)) {
133                                // internally, subscriptions that cannot activate will be set to error
134                                if (mySubscriptionActivatingInterceptor.activateSubscriptionIfRequired(theSubscription)) {
135                                        successfullyActivated = true;
136                                } else {
137                                        logSubscriptionNotActivatedPlusErrorIfPossible(theSubscription);
138                                }
139                        } else {
140                                ourLog.debug(
141                                                "Could not activate subscription {} because channel type {} is not supported.",
142                                                theSubscription.getIdElement(),
143                                                mySubscriptionCanonicalizer.getChannelType(theSubscription));
144                        }
145                }
146
147                return successfullyActivated;
148        }
149
150        /**
151         * Logs
152         *
153         * @param theSubscription
154         */
155        private void logSubscriptionNotActivatedPlusErrorIfPossible(IBaseResource theSubscription) {
156                String error;
157                if (theSubscription instanceof Subscription) {
158                        error = ((Subscription) theSubscription).getError();
159                } else if (theSubscription instanceof org.hl7.fhir.dstu3.model.Subscription) {
160                        error = ((org.hl7.fhir.dstu3.model.Subscription) theSubscription).getError();
161                } else if (theSubscription instanceof org.hl7.fhir.dstu2.model.Subscription) {
162                        error = ((org.hl7.fhir.dstu2.model.Subscription) theSubscription).getError();
163                } else {
164                        error = "";
165                }
166                ourLog.error(
167                                "Subscription {} could not be activated. "
168                                                + "This will not prevent startup, but it could lead to undesirable outcomes! {}",
169                                theSubscription.getIdElement().getIdPart(),
170                                (StringUtils.isBlank(error) ? "" : "Error: " + error));
171        }
172
173        public void syncSubscriptions() {
174                super.syncDatabaseToCache();
175        }
176}