001
014
015 package com.liferay.portal.kernel.notifications;
016
017 import com.liferay.portal.kernel.log.Log;
018 import com.liferay.portal.kernel.log.LogFactoryUtil;
019
020 import java.util.ArrayList;
021 import java.util.List;
022 import java.util.concurrent.locks.Lock;
023 import java.util.concurrent.locks.ReentrantLock;
024
025
028 public abstract class BaseChannelImpl implements Channel {
029
030 @Override
031 public void cleanUp() throws ChannelException {
032 lock.lock();
033
034 try {
035 long currentTime = System.currentTimeMillis();
036
037 if (currentTime > _nextCleanUpTime) {
038 _nextCleanUpTime = currentTime + _cleanUpInterval;
039
040 try {
041 doCleanUp();
042 }
043 catch (ChannelException ce) {
044 throw ce;
045 }
046 catch (Exception e) {
047 throw new ChannelException(e);
048 }
049 }
050 }
051 finally {
052 lock.unlock();
053 }
054 }
055
056 @Override
057 public void close() throws ChannelException {
058 flush();
059 }
060
061 public long getCompanyId() {
062 return _companyId;
063 }
064
065 @Override
066 public List<NotificationEvent> getNotificationEvents()
067 throws ChannelException {
068
069 return getNotificationEvents(true);
070 }
071
072 @Override
073 public long getUserId() {
074 return _userId;
075 }
076
077 public boolean hasNotificationEvents() {
078 try {
079 List<NotificationEvent> notificationEvents = getNotificationEvents(
080 false);
081
082 if (!notificationEvents.isEmpty()) {
083 return true;
084 }
085 }
086 catch (ChannelException ce) {
087 _log.error("Unable to fetch notifications", ce);
088 }
089
090 return false;
091 }
092
093 @Override
094 public void registerChannelListener(ChannelListener channelListener) {
095 lock.lock();
096
097 try {
098 List<ChannelListener> channelListeners = _getChannelListeners();
099
100 channelListeners.add(channelListener);
101
102 if (hasNotificationEvents()) {
103 notifyChannelListeners();
104 }
105 }
106 finally {
107 lock.unlock();
108 }
109 }
110
111 public void setCleanUpInterval(long cleanUpInterval) {
112 _cleanUpInterval = cleanUpInterval;
113 }
114
115 @Override
116 public void unregisterChannelListener(ChannelListener channelListener) {
117 lock.lock();
118
119 try {
120 List<ChannelListener> channelListeners = _getChannelListeners();
121
122 channelListeners.remove(channelListener);
123 }
124 finally {
125 lock.unlock();
126 }
127
128 channelListener.channelListenerRemoved(_userId);
129 }
130
131 protected BaseChannelImpl(long companyId, long usedId) {
132 _companyId = companyId;
133 _userId = usedId;
134 }
135
136 protected abstract void doCleanUp() throws Exception;
137
138 protected void notifyChannelListeners() {
139 for (ChannelListener channelListener : _getChannelListeners()) {
140 channelListener.notificationEventsAvailable(_userId);
141 }
142 }
143
144 protected final Lock lock = new ReentrantLock();
145
146 private List<ChannelListener> _getChannelListeners() {
147 if (_channelListeners == null) {
148 _channelListeners = new ArrayList<>();
149 }
150
151 return _channelListeners;
152 }
153
154 private static final Log _log = LogFactoryUtil.getLog(
155 BaseChannelImpl.class);
156
157 private List<ChannelListener> _channelListeners;
158 private long _cleanUpInterval;
159 private final long _companyId;
160 private long _nextCleanUpTime;
161 private final long _userId;
162
163 }