001
014
015 package com.liferay.portal.notifications;
016
017 import com.liferay.portal.kernel.exception.SystemException;
018 import com.liferay.portal.kernel.json.JSONException;
019 import com.liferay.portal.kernel.json.JSONFactoryUtil;
020 import com.liferay.portal.kernel.json.JSONObject;
021 import com.liferay.portal.kernel.log.Log;
022 import com.liferay.portal.kernel.log.LogFactoryUtil;
023 import com.liferay.portal.kernel.notifications.BaseChannelImpl;
024 import com.liferay.portal.kernel.notifications.Channel;
025 import com.liferay.portal.kernel.notifications.ChannelException;
026 import com.liferay.portal.kernel.notifications.NotificationEvent;
027 import com.liferay.portal.kernel.notifications.NotificationEventComparator;
028 import com.liferay.portal.kernel.notifications.NotificationEventFactoryUtil;
029 import com.liferay.portal.model.CompanyConstants;
030 import com.liferay.portal.model.UserNotificationEvent;
031 import com.liferay.portal.service.UserNotificationEventLocalServiceUtil;
032 import com.liferay.portal.util.PropsValues;
033
034 import java.util.ArrayList;
035 import java.util.Collection;
036 import java.util.Comparator;
037 import java.util.HashMap;
038 import java.util.HashSet;
039 import java.util.Iterator;
040 import java.util.List;
041 import java.util.Map;
042 import java.util.Set;
043 import java.util.TreeSet;
044 import java.util.concurrent.locks.ReentrantLock;
045
046
051 public class ChannelImpl extends BaseChannelImpl {
052
053 public ChannelImpl() {
054 this(CompanyConstants.SYSTEM, 0);
055 }
056
057 public ChannelImpl(long companyId, long usedId) {
058 super(companyId, usedId);
059 }
060
061 public Channel clone(long companyId, long userId) {
062 return new ChannelImpl(companyId, userId);
063 }
064
065 public void confirmDelivery(Collection<String> notificationEventUuids)
066 throws ChannelException {
067
068 confirmDelivery(notificationEventUuids, false);
069 }
070
071 public void confirmDelivery(
072 Collection<String> notificationEventUuids, boolean archive)
073 throws ChannelException {
074
075 _reentrantLock.lock();
076
077 try {
078 for (String notificationEventUuid : notificationEventUuids) {
079 Map<String, NotificationEvent> unconfirmedNotificationEvents =
080 _getUnconfirmedNotificationEvents();
081
082 unconfirmedNotificationEvents.remove(notificationEventUuid);
083 }
084
085 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED) {
086 if (archive) {
087 UserNotificationEventLocalServiceUtil.
088 updateUserNotificationEvents(
089 notificationEventUuids, archive);
090 }
091 else {
092 UserNotificationEventLocalServiceUtil.
093 deleteUserNotificationEvents(notificationEventUuids);
094 }
095 }
096 }
097 catch (Exception e) {
098 throw new ChannelException(
099 "Unable to confirm delivery for user " + getUserId() , e);
100 }
101 finally {
102 _reentrantLock.unlock();
103 }
104 }
105
106 public void confirmDelivery(String notificationEventUuid)
107 throws ChannelException {
108
109 confirmDelivery(notificationEventUuid, false);
110 }
111
112 public void confirmDelivery(String notificationEventUuid, boolean archive)
113 throws ChannelException {
114
115 _reentrantLock.lock();
116
117 try {
118 Map<String, NotificationEvent> unconfirmedNotificationEvents =
119 _getUnconfirmedNotificationEvents();
120
121 unconfirmedNotificationEvents.remove(notificationEventUuid);
122
123 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED) {
124 if (archive) {
125 UserNotificationEventLocalServiceUtil.
126 updateUserNotificationEvent(
127 notificationEventUuid, archive);
128 }
129 else {
130 UserNotificationEventLocalServiceUtil.
131 deleteUserNotificationEvent(notificationEventUuid);
132 }
133 }
134 }
135 catch (Exception e) {
136 throw new ChannelException(
137 "Uanble to confirm delivery for " + notificationEventUuid , e);
138 }
139 finally {
140 _reentrantLock.unlock();
141 }
142 }
143
144 public void deleteUserNotificiationEvent (String notificationEventUuid)
145 throws ChannelException {
146
147 try {
148 UserNotificationEventLocalServiceUtil.
149 deleteUserNotificationEvent(notificationEventUuid);
150 }
151 catch (Exception e) {
152 throw new ChannelException(
153 "Uanble to delete event " + notificationEventUuid ,
154 e);
155 }
156 }
157
158 public void deleteUserNotificiationEvents (
159 Collection<String> notificationEventUuids)
160 throws ChannelException {
161
162 try {
163 UserNotificationEventLocalServiceUtil.
164 deleteUserNotificationEvents(notificationEventUuids);
165 }
166 catch (Exception e) {
167 throw new ChannelException(
168 "Uanble to delete events for user " + getUserId() , e);
169 }
170 }
171
172 public void flush() {
173 _reentrantLock.lock();
174
175 try {
176 TreeSet<NotificationEvent> notificationEvents =
177 _getNotificationEvents();
178
179 notificationEvents.clear();
180 }
181 finally {
182 _reentrantLock.unlock();
183 }
184 }
185
186 public void flush(long timestamp) {
187 _reentrantLock.lock();
188
189 try {
190 TreeSet<NotificationEvent> notificationEvents =
191 _getNotificationEvents();
192
193 Iterator<NotificationEvent> itr = notificationEvents.iterator();
194
195 while (itr.hasNext()) {
196 NotificationEvent notificationEvent = itr.next();
197
198 if (notificationEvent.getTimestamp() < timestamp) {
199 itr.remove();
200 }
201 }
202 }
203 finally {
204 _reentrantLock.unlock();
205 }
206 }
207
208 public List<NotificationEvent> getNotificationEvents(boolean flush)
209 throws ChannelException {
210
211 _reentrantLock.lock();
212
213 try {
214 return doGetNotificationEvents(flush);
215 }
216 catch (ChannelException ce) {
217 throw ce;
218 }
219 catch (Exception e) {
220 throw new ChannelException(e);
221 }
222 finally {
223 _reentrantLock.unlock();
224 }
225 }
226
227 public void init() throws ChannelException {
228 _reentrantLock.lock();
229
230 try {
231 doInit();
232 }
233 catch (SystemException se) {
234 throw new ChannelException(
235 "Unable to init channel " + getUserId(), se);
236 }
237 finally {
238 _reentrantLock.unlock();
239 }
240 }
241
242 public void removeTransientNotificationEvents(
243 Collection<NotificationEvent> notificationEvents) {
244
245 _reentrantLock.lock();
246
247 try {
248 TreeSet<NotificationEvent> notificationEventsSet =
249 _getNotificationEvents();
250
251 notificationEventsSet.removeAll(notificationEvents);
252 }
253 finally {
254 _reentrantLock.unlock();
255 }
256 }
257
258 public void removeTransientNotificationEventsByUuid(
259 Collection<String> notificationEventUuids) {
260
261 Set<String> notificationEventUuidsSet = new HashSet<String>(
262 notificationEventUuids);
263
264 _reentrantLock.lock();
265
266 try {
267 TreeSet<NotificationEvent> notificationEvents =
268 _getNotificationEvents();
269
270 Iterator<NotificationEvent> itr = notificationEvents.iterator();
271
272 while (itr.hasNext()) {
273 NotificationEvent notificationEvent = itr.next();
274
275 if (notificationEventUuidsSet.contains(
276 notificationEvent.getUuid())) {
277
278 itr.remove();
279 }
280 }
281 }
282 finally {
283 _reentrantLock.unlock();
284 }
285 }
286
287 public void sendNotificationEvent(NotificationEvent notificationEvent)
288 throws ChannelException {
289
290 _reentrantLock.lock();
291
292 try {
293 long currentTime = System.currentTimeMillis();
294
295 storeNotificationEvent(notificationEvent, currentTime);
296
297 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
298 notificationEvent.isDeliveryRequired()) {
299
300 UserNotificationEventLocalServiceUtil.addUserNotificationEvent(
301 getUserId(), notificationEvent);
302 }
303 }
304 catch (Exception e) {
305 throw new ChannelException("Unable to send event", e);
306 }
307 finally {
308 _reentrantLock.unlock();
309 }
310
311 notifyChannelListeners();
312 }
313
314 public void sendNotificationEvents(
315 Collection<NotificationEvent> notificationEvents)
316 throws ChannelException {
317
318 _reentrantLock.lock();
319
320 try {
321 long currentTime = System.currentTimeMillis();
322
323 List<NotificationEvent> persistedNotificationEvents =
324 new ArrayList<NotificationEvent>(notificationEvents.size());
325
326 for (NotificationEvent notificationEvent : notificationEvents) {
327 storeNotificationEvent(notificationEvent, currentTime);
328
329 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
330 notificationEvent.isDeliveryRequired()) {
331
332 persistedNotificationEvents.add(notificationEvent);
333 }
334 }
335
336 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
337 !persistedNotificationEvents.isEmpty()) {
338
339 UserNotificationEventLocalServiceUtil.addUserNotificationEvents(
340 getUserId(), persistedNotificationEvents);
341 }
342 }
343 catch (Exception e) {
344 throw new ChannelException("Unable to send event", e);
345 }
346 finally {
347 _reentrantLock.unlock();
348 }
349
350 notifyChannelListeners();
351 }
352
353 @Override
354 protected void doCleanUp() throws Exception {
355 _reentrantLock.lock();
356
357 try {
358 long currentTime = System.currentTimeMillis();
359
360 TreeSet<NotificationEvent> notificationEvents =
361 _getNotificationEvents();
362
363 Iterator<NotificationEvent> itr1 = notificationEvents.iterator();
364
365 while (itr1.hasNext()) {
366 NotificationEvent notificationEvent = itr1.next();
367
368 if (isRemoveNotificationEvent(
369 notificationEvent, currentTime)) {
370
371 itr1.remove();
372 }
373 }
374
375 Map<String, NotificationEvent> unconfirmedNotificationEvents =
376 _getUnconfirmedNotificationEvents();
377
378 List<String> invalidNotificationEventUuids = new ArrayList<String>(
379 unconfirmedNotificationEvents.size());
380
381 Set<Map.Entry<String, NotificationEvent>>
382 unconfirmedNotificationEventsSet =
383 unconfirmedNotificationEvents.entrySet();
384
385 Iterator<Map.Entry<String, NotificationEvent>> itr2 =
386 unconfirmedNotificationEventsSet.iterator();
387
388 while (itr2.hasNext()) {
389 Map.Entry<String, NotificationEvent> entry =
390 itr2.next();
391
392 NotificationEvent notificationEvent = entry.getValue();
393
394 if (isRemoveNotificationEvent(notificationEvent, currentTime)) {
395 invalidNotificationEventUuids.add(entry.getKey());
396
397 itr2.remove();
398 }
399 }
400
401 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
402 !invalidNotificationEventUuids.isEmpty()) {
403
404 UserNotificationEventLocalServiceUtil.
405 deleteUserNotificationEvents(invalidNotificationEventUuids);
406 }
407 }
408 catch (Exception e) {
409 throw new ChannelException(
410 "Unable to clean up channel " + getUserId(), e);
411 }
412 finally {
413 _reentrantLock.unlock();
414 }
415 }
416
417 protected List<NotificationEvent> doGetNotificationEvents(boolean flush)
418 throws Exception {
419
420 long currentTime = System.currentTimeMillis();
421
422 TreeSet<NotificationEvent> notificationEventsSet =
423 _getNotificationEvents();
424
425 Map<String, NotificationEvent> unconfirmedNotificationEvents =
426 _getUnconfirmedNotificationEvents();
427
428 List<NotificationEvent> notificationEvents =
429 new ArrayList<NotificationEvent>(
430 notificationEventsSet.size() +
431 unconfirmedNotificationEvents.size());
432
433 for (NotificationEvent notificationEvent : notificationEventsSet) {
434 if (isRemoveNotificationEvent(notificationEvent, currentTime)) {
435 break;
436 }
437 else {
438 notificationEvents.add(notificationEvent);
439 }
440 }
441
442 if (flush) {
443 notificationEventsSet.clear();
444 }
445 else if (notificationEventsSet.size() != notificationEvents.size()) {
446 notificationEventsSet.retainAll(notificationEvents);
447 }
448
449 List<String> invalidNotificationEventUuids = new ArrayList<String>(
450 unconfirmedNotificationEvents.size());
451
452 Set<Map.Entry<String, NotificationEvent>>
453 unconfirmedNotificationEventsSet =
454 unconfirmedNotificationEvents.entrySet();
455
456 Iterator<Map.Entry<String, NotificationEvent>> itr =
457 unconfirmedNotificationEventsSet.iterator();
458
459 while (itr.hasNext()) {
460 Map.Entry<String, NotificationEvent> entry = itr.next();
461
462 NotificationEvent notificationEvent = entry.getValue();
463
464 if (isRemoveNotificationEvent(notificationEvent, currentTime) &&
465 !notificationEvent.isArchived()) {
466
467 invalidNotificationEventUuids.add(notificationEvent.getUuid());
468
469 itr.remove();
470 }
471 else {
472 notificationEvents.add(entry.getValue());
473 }
474 }
475
476 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
477 !invalidNotificationEventUuids.isEmpty()) {
478
479 UserNotificationEventLocalServiceUtil.deleteUserNotificationEvents(
480 invalidNotificationEventUuids);
481 }
482
483 return notificationEvents;
484 }
485
486 protected void doInit() throws SystemException {
487 if (!PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED) {
488 return;
489 }
490
491 List<UserNotificationEvent> userNotificationEvents =
492 UserNotificationEventLocalServiceUtil.getUserNotificationEvents(
493 getUserId(), false);
494
495 Map<String, NotificationEvent> unconfirmedNotificationEvents =
496 _getUnconfirmedNotificationEvents();
497
498 List<String> invalidNotificationEventUuids = new ArrayList<String>(
499 unconfirmedNotificationEvents.size());
500
501 long currentTime = System.currentTimeMillis();
502
503 for (UserNotificationEvent persistedNotificationEvent :
504 userNotificationEvents) {
505
506 try {
507 JSONObject payloadJSONObject = JSONFactoryUtil.createJSONObject(
508 persistedNotificationEvent.getPayload());
509
510 NotificationEvent notificationEvent =
511 NotificationEventFactoryUtil.createNotificationEvent(
512 persistedNotificationEvent.getTimestamp(),
513 persistedNotificationEvent.getType(),
514 payloadJSONObject);
515
516 notificationEvent.setDeliveryRequired(
517 persistedNotificationEvent.getDeliverBy());
518
519 notificationEvent.setUuid(persistedNotificationEvent.getUuid());
520
521 if (isRemoveNotificationEvent(notificationEvent, currentTime)) {
522 invalidNotificationEventUuids.add(
523 notificationEvent.getUuid());
524 }
525 else {
526 unconfirmedNotificationEvents.put(
527 notificationEvent.getUuid(), notificationEvent);
528 }
529 }
530 catch (JSONException jsone) {
531 _log.error(jsone, jsone);
532
533 invalidNotificationEventUuids.add(
534 persistedNotificationEvent.getUuid());
535 }
536 }
537
538 if (!invalidNotificationEventUuids.isEmpty()) {
539 UserNotificationEventLocalServiceUtil.deleteUserNotificationEvents(
540 invalidNotificationEventUuids);
541 }
542 }
543
544 protected boolean isRemoveNotificationEvent(
545 NotificationEvent notificationEvent, long currentTime) {
546
547 if ((notificationEvent.getDeliverBy() != 0) &&
548 (notificationEvent.getDeliverBy() <= currentTime)) {
549
550 return true;
551 }
552 else {
553 return false;
554 }
555 }
556
557 protected void storeNotificationEvent(
558 NotificationEvent notificationEvent, long currentTime) {
559
560 if (isRemoveNotificationEvent(notificationEvent, currentTime)) {
561 return;
562 }
563
564 if (PropsValues.USER_NOTIFICATION_EVENT_CONFIRMATION_ENABLED &&
565 notificationEvent.isDeliveryRequired()) {
566
567 Map<String, NotificationEvent> unconfirmedNotificationEvents =
568 _getUnconfirmedNotificationEvents();
569
570 unconfirmedNotificationEvents.put(
571 notificationEvent.getUuid(), notificationEvent);
572 }
573 else {
574 TreeSet<NotificationEvent> notificationEvents =
575 _getNotificationEvents();
576
577 notificationEvents.add(notificationEvent);
578
579 if (notificationEvents.size() >
580 PropsValues.NOTIFICATIONS_MAX_EVENTS) {
581
582 NotificationEvent firstNotificationEvent =
583 notificationEvents.first();
584
585 notificationEvents.remove(firstNotificationEvent);
586 }
587 }
588 }
589
590 private TreeSet<NotificationEvent> _getNotificationEvents() {
591 if (_notificationEvents == null) {
592 _notificationEvents = new TreeSet<NotificationEvent>(_comparator);
593 }
594
595 return _notificationEvents;
596 }
597
598 private Map<String, NotificationEvent> _getUnconfirmedNotificationEvents() {
599 if (_unconfirmedNotificationEvents == null) {
600 _unconfirmedNotificationEvents =
601 new HashMap<String, NotificationEvent>();
602 }
603
604 return _unconfirmedNotificationEvents;
605 }
606
607 private static Log _log = LogFactoryUtil.getLog(ChannelImpl.class);
608
609 private static Comparator<NotificationEvent> _comparator =
610 new NotificationEventComparator();
611
612 private TreeSet<NotificationEvent> _notificationEvents;
613 private ReentrantLock _reentrantLock = new ReentrantLock();
614 private Map<String, NotificationEvent> _unconfirmedNotificationEvents;
615
616 }