View Javadoc
1   /*
2    * Copyright 2016 the original author or authors.
3    *
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    *
8    *       http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  package org.openehealth.ipf.boot.xds;
17  
18  import org.openehealth.ipf.commons.ihe.ws.correlation.AsynchronyCorrelator;
19  import org.openehealth.ipf.commons.ihe.ws.cxf.audit.WsAuditDataset;
20  import org.springframework.cache.Cache;
21  import org.springframework.cache.CacheManager;
22  
23  import static java.util.Objects.requireNonNull;
24  
25  /**
26   * Ehcache-based implementation of asynchronous message correlator.
27   *
28   * @author Dmytro Rud
29   */
30  public class CachingAsynchronyCorrelator<AuditDatasetType extends WsAuditDataset> implements AsynchronyCorrelator<AuditDatasetType> {
31  
32      private static final String ASYNCHRONY_CORRELATOR_CACHE = "asynchronyCorrelatorCache";
33      private static final String SERVICE_ENDPOINT_URI_SUFFIX = ".serviceEndpoint";
34      private static final String CORRELATION_KEY_SUFFIX = ".correlationKey";
35      private static final String AUDIT_DATASET_SUFFIX = ".auditDataset";
36      private static final String ALTERNATIVE_KEY_SUFFIX = ".alternativeKey";
37      private static final String ALTERNATIVE_KEYS_SUFFIX = ".alternativeKeys";
38  
39      private final Cache cache;
40  
41      public CachingAsynchronyCorrelator(CacheManager cacheManager) {
42          this.cache = cacheManager.getCache(ASYNCHRONY_CORRELATOR_CACHE);
43      }
44  
45      @Override
46      public void storeServiceEndpointUri(String messageId, String serviceEndpointUri) {
47          cache.put(messageId + SERVICE_ENDPOINT_URI_SUFFIX, requireNonNull(serviceEndpointUri, "service endpoint URI"));
48      }
49  
50      @Override
51      public void storeCorrelationKey(String messageId, String correlationKey) {
52          cache.put(messageId + CORRELATION_KEY_SUFFIX, requireNonNull(correlationKey, "correlation key"));
53      }
54  
55      @Override
56      public void storeAuditDataset(String messageId, WsAuditDataset auditDataset) {
57          cache.put(messageId + AUDIT_DATASET_SUFFIX, requireNonNull(auditDataset, "audit dataset"));
58      }
59  
60      @Override
61      public String getServiceEndpointUri(String messageId) {
62          return cache.get(messageId + SERVICE_ENDPOINT_URI_SUFFIX, String.class);
63      }
64  
65      @Override
66      public String getCorrelationKey(String messageId) {
67          return cache.get(messageId + CORRELATION_KEY_SUFFIX, String.class);
68      }
69  
70      @Override
71      public AuditDatasetType getAuditDataset(String messageId) {
72          return (AuditDatasetType) cache.get(messageId + AUDIT_DATASET_SUFFIX).get();
73      }
74  
75      @Override
76      public void storeAlternativeKeys(String messageId, String... alternativeKeys) {
77          requireNonNull(alternativeKeys, "alternative keys should be not null");
78          for (String key : alternativeKeys) {
79              cache.put(key + ALTERNATIVE_KEY_SUFFIX, messageId);
80          }
81          cache.put(messageId + ALTERNATIVE_KEYS_SUFFIX, alternativeKeys);
82      }
83  
84      @Override
85      public String getMessageId(String alternativeKey) {
86          return cache.get(alternativeKey + ALTERNATIVE_KEY_SUFFIX, String.class);
87      }
88  
89      @Override
90      public boolean delete(String messageId) {
91          String[] alternativeKeys = cache.get(messageId + ALTERNATIVE_KEYS_SUFFIX, String[].class);
92          if (alternativeKeys != null) {
93              for (String key : alternativeKeys) {
94                  cache.evict(key + ALTERNATIVE_KEY_SUFFIX);
95              }
96          }
97          cache.evict(messageId + ALTERNATIVE_KEYS_SUFFIX);
98          cache.evict(messageId + CORRELATION_KEY_SUFFIX);
99          cache.evict(messageId + AUDIT_DATASET_SUFFIX);
100         if (cache.get(messageId + SERVICE_ENDPOINT_URI_SUFFIX) != null) {
101             cache.evict(messageId + SERVICE_ENDPOINT_URI_SUFFIX);
102             return true;
103         }
104         return false;
105     }
106 }