1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package org.kuali.rice.kim.service.impl;
17
18 import org.apache.commons.lang.StringUtils;
19 import org.apache.log4j.Level;
20 import org.apache.log4j.Logger;
21 import org.kuali.rice.core.api.config.property.ConfigurationService;
22 import org.kuali.rice.kim.api.KimConstants;
23 import org.kuali.rice.kim.impl.identity.IdentityArchiveService;
24 import org.kuali.rice.kim.api.identity.entity.EntityDefault;
25 import org.kuali.rice.kim.api.identity.principal.Principal;
26 import org.kuali.rice.kim.impl.identity.EntityDefaultInfoCacheBo;
27 import org.kuali.rice.krad.service.BusinessObjectService;
28 import org.kuali.rice.krad.service.KRADServiceLocatorInternal;
29 import org.kuali.rice.ksb.service.KSBServiceLocator;
30 import org.springframework.beans.factory.DisposableBean;
31 import org.springframework.beans.factory.InitializingBean;
32 import org.springframework.transaction.PlatformTransactionManager;
33 import org.springframework.transaction.TransactionStatus;
34 import org.springframework.transaction.support.TransactionCallback;
35 import org.springframework.transaction.support.TransactionTemplate;
36
37 import java.util.ArrayList;
38 import java.util.Arrays;
39 import java.util.Collection;
40 import java.util.Collections;
41 import java.util.Comparator;
42 import java.util.HashMap;
43 import java.util.HashSet;
44 import java.util.List;
45 import java.util.Map;
46 import java.util.Set;
47 import java.util.concurrent.Callable;
48 import java.util.concurrent.ConcurrentLinkedQueue;
49 import java.util.concurrent.TimeUnit;
50 import java.util.concurrent.atomic.AtomicBoolean;
51 import java.util.concurrent.atomic.AtomicInteger;
52
53
54
55
56
57
58
59 public class IdentityArchiveServiceImpl implements IdentityArchiveService, InitializingBean, DisposableBean {
60 private static final Logger LOG = Logger.getLogger( IdentityArchiveServiceImpl.class );
61
62 private BusinessObjectService businessObjectService;
63 private ConfigurationService kualiConfigurationService;
64
65 private static final String EXEC_INTERVAL_SECS = "kim.identityArchiveServiceImpl.executionIntervalSeconds";
66 private static final String MAX_WRITE_QUEUE_SIZE = "kim.identityArchiveServiceImpl.maxWriteQueueSize";
67 private static final int EXECUTION_INTERVAL_SECONDS_DEFAULT = 600;
68 private static final int MAX_WRITE_QUEUE_SIZE_DEFAULT = 300;
69
70 private final WriteQueue writeQueue = new WriteQueue();
71 private final EntityArchiveWriter writer = new EntityArchiveWriter();
72
73
74 private final Runnable maxQueueSizeExceededWriter =
75 new CallableAdapter(new PreLogCallableWrapper<Boolean>(writer, Level.DEBUG, "max size exceeded, flushing write queue"));
76
77
78 private final Runnable scheduledWriter =
79 new CallableAdapter(new PreLogCallableWrapper<Boolean>(writer, Level.DEBUG, "scheduled write out, flushing write queue"));
80
81
82 private final Runnable shutdownWriter =
83 new CallableAdapter(new PreLogCallableWrapper<Boolean>(writer, Level.DEBUG, "rice is shutting down, flushing write queue"));
84
85 private int getExecutionIntervalSeconds() {
86 final String prop = kualiConfigurationService.getPropertyValueAsString(EXEC_INTERVAL_SECS);
87 try {
88 return Integer.valueOf(prop).intValue();
89 } catch (NumberFormatException e) {
90 return EXECUTION_INTERVAL_SECONDS_DEFAULT;
91 }
92 }
93
94 private int getMaxWriteQueueSize() {
95 final String prop = kualiConfigurationService.getPropertyValueAsString(MAX_WRITE_QUEUE_SIZE);
96 try {
97 return Integer.valueOf(prop).intValue();
98 } catch (NumberFormatException e) {
99 return MAX_WRITE_QUEUE_SIZE_DEFAULT;
100 }
101 }
102
103 @Override
104 public EntityDefault getEntityDefaultFromArchive( String entityId ) {
105 if (StringUtils.isBlank(entityId)) {
106 throw new IllegalArgumentException("entityId is blank");
107 }
108
109 Map<String,String> criteria = new HashMap<String, String>(1);
110 criteria.put(KimConstants.PrimaryKeyConstants.SUB_ENTITY_ID, entityId);
111 EntityDefaultInfoCacheBo cachedValue = businessObjectService.findByPrimaryKey(EntityDefaultInfoCacheBo.class, criteria);
112 return (cachedValue == null) ? null : cachedValue.convertCacheToEntityDefaultInfo();
113 }
114
115 @Override
116 public EntityDefault getEntityDefaultFromArchiveByPrincipalId(String principalId) {
117 if (StringUtils.isBlank(principalId)) {
118 throw new IllegalArgumentException("principalId is blank");
119 }
120
121 Map<String,String> criteria = new HashMap<String, String>(1);
122 criteria.put("principalId", principalId);
123 EntityDefaultInfoCacheBo cachedValue = businessObjectService.findByPrimaryKey(EntityDefaultInfoCacheBo.class, criteria);
124 return (cachedValue == null) ? null : cachedValue.convertCacheToEntityDefaultInfo();
125 }
126
127 @Override
128 public EntityDefault getEntityDefaultFromArchiveByPrincipalName(String principalName) {
129 if (StringUtils.isBlank(principalName)) {
130 throw new IllegalArgumentException("principalName is blank");
131 }
132
133 Map<String,String> criteria = new HashMap<String, String>(1);
134 criteria.put("principalName", principalName);
135 Collection<EntityDefaultInfoCacheBo> entities = businessObjectService.findMatching(EntityDefaultInfoCacheBo.class, criteria);
136 return (entities == null || entities.isEmpty()) ? null : entities.iterator().next().convertCacheToEntityDefaultInfo();
137 }
138
139 @Override
140 public void saveEntityDefaultToArchive(EntityDefault entity) {
141 if (entity == null) {
142 throw new IllegalArgumentException("entity is blank");
143 }
144
145
146 if (getMaxWriteQueueSize() <= writeQueue.offerAndGetSize(entity)
147 writer.requestSubmit()) {
148 KSBServiceLocator.getThreadPool().execute(maxQueueSizeExceededWriter);
149 }
150 }
151
152 public void setBusinessObjectService(BusinessObjectService businessObjectService) {
153 this.businessObjectService = businessObjectService;
154 }
155
156 public void setKualiConfigurationService(
157 ConfigurationService kualiConfigurationService) {
158 this.kualiConfigurationService = kualiConfigurationService;
159 }
160
161
162 @Override
163 public void afterPropertiesSet() throws Exception {
164 LOG.info("scheduling writer...");
165 KSBServiceLocator.getScheduledPool().scheduleAtFixedRate(scheduledWriter,
166 getExecutionIntervalSeconds(), getExecutionIntervalSeconds(), TimeUnit.SECONDS);
167 }
168
169
170 @Override
171 public void destroy() throws Exception {
172 KSBServiceLocator.getThreadPool().execute(shutdownWriter);
173 }
174
175
176
177
178
179
180
181
182 private class EntityArchiveWriter implements Callable {
183
184
185 AtomicBoolean currentlySubmitted = new AtomicBoolean(false);
186
187 private final Comparator<Comparable> nullSafeComparator = new Comparator<Comparable>() {
188 @Override
189 public int compare(Comparable i1, Comparable i2) {
190 if (i1 != null && i2 != null) {
191 return i1.compareTo(i2);
192 } else if (i1 == null) {
193 if (i2 == null) {
194 return 0;
195 } else {
196 return -1;
197 }
198 } else {
199 return 1;
200 }
201 };
202 };
203
204
205
206
207 private final Comparator<EntityDefault> kediComparator = new Comparator<EntityDefault>() {
208
209
210
211
212 @Override
213 public int compare(EntityDefault o1, EntityDefault o2) {
214 String entityId1 = (o1 == null) ? null : o1.getEntityId();
215 String entityId2 = (o2 == null) ? null : o2.getEntityId();
216
217 int result = nullSafeComparator.compare(entityId1, entityId2);
218
219 if (result == 0) {
220 result = getPrincipalIdsString(o1).compareTo(getPrincipalIdsString(o2));
221 }
222
223 return result;
224 }
225
226
227
228
229
230
231
232 private String getPrincipalIdsString(EntityDefault entity) {
233 String result = "";
234 if (entity != null) {
235 List<Principal> principals = entity.getPrincipals();
236 if (principals != null) {
237 if (principals.size() == 1) {
238 result = principals.get(0).getPrincipalId();
239 } else {
240 String [] ids = new String [principals.size()];
241 int insertIndex = 0;
242 for (Principal principal : principals) {
243 ids[insertIndex++] = principal.getPrincipalId();
244 }
245 Arrays.sort(ids);
246 result = StringUtils.join(ids, "\n");
247 }
248 }
249 }
250 return result;
251 }
252 };
253
254 public boolean requestSubmit() {
255 return currentlySubmitted.compareAndSet(false, true);
256 }
257
258
259
260
261
262 @Override
263 public Object call() {
264 try {
265
266
267
268 PlatformTransactionManager transactionManager = KRADServiceLocatorInternal.getTransactionManager();
269 TransactionTemplate template = new TransactionTemplate(transactionManager);
270 template.execute(new TransactionCallback() {
271 @Override
272 public Object doInTransaction(TransactionStatus status) {
273 EntityDefault entity = null;
274 ArrayList<EntityDefault> entitiesToInsert = new ArrayList<EntityDefault>(getMaxWriteQueueSize());
275 Set<String> deduper = new HashSet<String>(getMaxWriteQueueSize());
276
277
278 while (entitiesToInsert.size() < getMaxWriteQueueSize() && null != (entity = writeQueue.poll())) {
279 if (deduper.add(entity.getEntityId())) {
280 entitiesToInsert.add(entity);
281 }
282 }
283
284 Collections.sort(entitiesToInsert, kediComparator);
285 List<EntityDefaultInfoCacheBo> entityCache = new ArrayList<EntityDefaultInfoCacheBo>(entitiesToInsert.size());
286 for (EntityDefault entityToInsert : entitiesToInsert) {
287 entityCache.add(new EntityDefaultInfoCacheBo( entityToInsert ));
288 }
289 businessObjectService.save(entityCache);
290
291
292
293 return null;
294 }
295 });
296 } finally {
297 currentlySubmitted.compareAndSet(true, false);
298 }
299
300 return Boolean.TRUE;
301 }
302 }
303
304
305
306
307
308
309
310
311
312 private static class WriteQueue {
313 AtomicInteger writeQueueSize = new AtomicInteger(0);
314 ConcurrentLinkedQueue<EntityDefault> queue = new ConcurrentLinkedQueue<EntityDefault>();
315
316 public int offerAndGetSize(EntityDefault entity) {
317 queue.add(entity);
318 return writeQueueSize.incrementAndGet();
319 }
320
321 private EntityDefault poll() {
322 EntityDefault result = queue.poll();
323 if (result != null) { writeQueueSize.decrementAndGet(); }
324 return result;
325 }
326 }
327
328
329
330
331
332
333
334 private static class PreLogCallableWrapper<A> implements Callable<A> {
335
336 private final Callable inner;
337 private final Level level;
338 private final String message;
339
340 public PreLogCallableWrapper(Callable inner, Level level, String message) {
341 this.inner = inner;
342 this.level = level;
343 this.message = message;
344 }
345
346
347
348
349
350
351 @Override
352 @SuppressWarnings("unchecked")
353 public A call() throws Exception {
354 LOG.log(level, message);
355 return (A)inner.call();
356 }
357 }
358
359
360
361
362
363
364
365 private static class CallableAdapter implements Runnable {
366
367 private final Callable callable;
368
369 public CallableAdapter(Callable callable) {
370 this.callable = callable;
371 }
372
373 @Override
374 public void run() {
375 try {
376 callable.call();
377 } catch (Exception e) {
378 throw new RuntimeException(e);
379 }
380 }
381 }
382 }