001 /*
002 GRANITE DATA SERVICES
003 Copyright (C) 2011 GRANITE DATA SERVICES S.A.S.
004
005 This file is part of Granite Data Services.
006
007 Granite Data Services is free software; you can redistribute it and/or modify
008 it under the terms of the GNU Library General Public License as published by
009 the Free Software Foundation; either version 2 of the License, or (at your
010 option) any later version.
011
012 Granite Data Services is distributed in the hope that it will be useful, but
013 WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
014 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License
015 for more details.
016
017 You should have received a copy of the GNU Library General Public License
018 along with this library; if not, see <http://www.gnu.org/licenses/>.
019 */
020
021 package org.granite.gravity;
022
023 import java.io.IOException;
024 import java.io.ObjectOutput;
025 import java.io.OutputStream;
026 import java.util.Collection;
027 import java.util.LinkedList;
028 import java.util.concurrent.ConcurrentHashMap;
029 import java.util.concurrent.ConcurrentMap;
030 import java.util.concurrent.locks.Lock;
031 import java.util.concurrent.locks.ReentrantLock;
032
033 import javax.servlet.http.HttpServletRequest;
034 import javax.servlet.http.HttpServletResponse;
035
036 import org.granite.context.AMFContextImpl;
037 import org.granite.context.GraniteContext;
038 import org.granite.logging.Logger;
039 import org.granite.messaging.webapp.HttpGraniteContext;
040 import org.granite.util.ContentType;
041
042 import flex.messaging.messages.AsyncMessage;
043
044 /**
045 * @author Franck WOLFF
046 */
047 public abstract class AbstractChannel implements Channel {
048
049 ///////////////////////////////////////////////////////////////////////////
050 // Fields.
051
052 private static final Logger log = Logger.getLogger(AbstractChannel.class);
053
054 protected final String id;
055 protected final String sessionId;
056 protected final String clientType;
057 protected final Gravity gravity;
058 protected final ChannelFactory<? extends Channel> factory;
059 // protected final ServletConfig servletConfig;
060
061 protected final ConcurrentMap<String, Subscription> subscriptions = new ConcurrentHashMap<String, Subscription>();
062
063 protected LinkedList<AsyncPublishedMessage> publishedQueue = new LinkedList<AsyncPublishedMessage>();
064 protected final Lock publishedQueueLock = new ReentrantLock();
065
066 protected LinkedList<AsyncMessage> receivedQueue = new LinkedList<AsyncMessage>();
067 protected final Lock receivedQueueLock = new ReentrantLock();
068
069 protected final AsyncPublisher publisher;
070 protected final AsyncReceiver receiver;
071
072 ///////////////////////////////////////////////////////////////////////////
073 // Constructor.
074
075 protected AbstractChannel(Gravity gravity, String id, ChannelFactory<? extends Channel> factory, String clientType) {
076 if (id == null)
077 throw new NullPointerException("id cannot be null");
078
079 this.id = id;
080 GraniteContext graniteContext = GraniteContext.getCurrentInstance();
081 this.clientType = clientType;
082 this.sessionId = graniteContext != null ? graniteContext.getSessionId() : null;
083 this.gravity = gravity;
084 this.factory = factory;
085
086 this.publisher = new AsyncPublisher(this);
087 this.receiver = new AsyncReceiver(this);
088 }
089
090 ///////////////////////////////////////////////////////////////////////////
091 // Abstract protected method.
092
093 protected abstract boolean hasAsyncHttpContext();
094 protected abstract AsyncHttpContext acquireAsyncHttpContext();
095 protected abstract void releaseAsyncHttpContext(AsyncHttpContext context);
096
097 ///////////////////////////////////////////////////////////////////////////
098 // Channel interface implementation.
099
100 public String getId() {
101 return id;
102 }
103
104 public String getClientType() {
105 return clientType;
106 }
107
108 public ChannelFactory<? extends Channel> getFactory() {
109 return factory;
110 }
111
112 public Gravity getGravity() {
113 return gravity;
114 }
115
116 public Subscription addSubscription(String destination, String subTopicId, String subscriptionId, boolean noLocal) {
117 Subscription subscription = new Subscription(this, destination, subTopicId, subscriptionId, noLocal);
118 Subscription present = subscriptions.putIfAbsent(subscriptionId, subscription);
119 return (present != null ? present : subscription);
120 }
121
122 public Collection<Subscription> getSubscriptions() {
123 return subscriptions.values();
124 }
125
126 public Subscription removeSubscription(String subscriptionId) {
127 return subscriptions.remove(subscriptionId);
128 }
129
130 public void publish(AsyncPublishedMessage message) throws MessagePublishingException {
131 if (message == null)
132 throw new NullPointerException("message cannot be null");
133
134 publishedQueueLock.lock();
135 try {
136 publishedQueue.add(message);
137 }
138 finally {
139 publishedQueueLock.unlock();
140 }
141
142 publisher.queue(getGravity());
143 }
144
145 public boolean hasPublishedMessage() {
146 publishedQueueLock.lock();
147 try {
148 return !publishedQueue.isEmpty();
149 }
150 finally {
151 publishedQueueLock.unlock();
152 }
153 }
154
155 public boolean runPublish() {
156 LinkedList<AsyncPublishedMessage> publishedCopy = null;
157
158 publishedQueueLock.lock();
159 try {
160 if (publishedQueue.isEmpty())
161 return false;
162 publishedCopy = publishedQueue;
163 publishedQueue = new LinkedList<AsyncPublishedMessage>();
164 }
165 finally {
166 publishedQueueLock.unlock();
167 }
168
169 for (AsyncPublishedMessage message : publishedCopy) {
170 try {
171 message.publish(this);
172 }
173 catch (Exception e) {
174 log.error(e, "Error while trying to publish message: %s", message);
175 }
176 }
177
178 return true;
179 }
180
181 public void receive(AsyncMessage message) throws MessageReceivingException {
182 if (message == null)
183 throw new NullPointerException("message cannot be null");
184
185 Gravity gravity = getGravity();
186
187 receivedQueueLock.lock();
188 try {
189 if (receivedQueue.size() + 1 > gravity.getGravityConfig().getMaxMessagesQueuedPerChannel())
190 throw new MessageReceivingException(message, "Could not queue message (channel's queue is full) for channel: " + this);
191
192 receivedQueue.add(message);
193 }
194 finally {
195 receivedQueueLock.unlock();
196 }
197
198 if (hasAsyncHttpContext())
199 receiver.queue(gravity);
200 }
201
202 public boolean hasReceivedMessage() {
203 receivedQueueLock.lock();
204 try {
205 return !receivedQueue.isEmpty();
206 }
207 finally {
208 receivedQueueLock.unlock();
209 }
210 }
211
212 public boolean runReceive() {
213 return runReceived(null);
214 }
215
216 protected ObjectOutput newSerializer(GraniteContext context, OutputStream os) {
217 return context.getGraniteConfig().newAMF3Serializer(os);
218 }
219
220 protected String getSerializerContentType() {
221 return ContentType.AMF.mimeType();
222 }
223
224 public boolean runReceived(AsyncHttpContext asyncHttpContext) {
225
226 boolean httpAsParam = (asyncHttpContext != null);
227 LinkedList<AsyncMessage> messages = null;
228 OutputStream os = null;
229
230 try {
231 receivedQueueLock.lock();
232 try {
233 // Do we have any pending messages?
234 if (receivedQueue.isEmpty())
235 return false;
236
237 // Do we have a valid http context?
238 if (asyncHttpContext == null) {
239 asyncHttpContext = acquireAsyncHttpContext();
240 if (asyncHttpContext == null)
241 return false;
242 }
243
244 // Both conditions are ok, get all pending messages.
245 messages = receivedQueue;
246 receivedQueue = new LinkedList<AsyncMessage>();
247 }
248 finally {
249 receivedQueueLock.unlock();
250 }
251
252 HttpServletRequest request = asyncHttpContext.getRequest();
253 HttpServletResponse response = asyncHttpContext.getResponse();
254
255 // Set response messages correlation ids to connect request message id.
256 String correlationId = asyncHttpContext.getConnectMessage().getMessageId();
257 AsyncMessage[] messagesArray = new AsyncMessage[messages.size()];
258 int i = 0;
259 for (AsyncMessage message : messages) {
260 message.setCorrelationId(correlationId);
261 messagesArray[i++] = message;
262 }
263
264 // Setup serialization context (thread local)
265 Gravity gravity = getGravity();
266 GraniteContext context = HttpGraniteContext.createThreadIntance(
267 gravity.getGraniteConfig(), gravity.getServicesConfig(),
268 null, request, response
269 );
270 ((AMFContextImpl)context.getAMFContext()).setCurrentAmf3Message(asyncHttpContext.getConnectMessage());
271
272 // Write messages to response output stream.
273
274 response.setStatus(HttpServletResponse.SC_OK);
275 response.setContentType(getSerializerContentType());
276 response.setDateHeader("Expire", 0L);
277 response.setHeader("Cache-Control", "no-store");
278
279 os = response.getOutputStream();
280 ObjectOutput serializer = newSerializer(context, os);
281
282 log.debug("<< [MESSAGES for channel=%s] %s", this, messagesArray);
283
284 serializer.writeObject(messagesArray);
285
286 os.flush();
287 response.flushBuffer();
288
289 return true; // Messages were delivered, http context isn't valid anymore.
290 }
291 catch (IOException e) {
292 log.warn(e, "Could not send messages to channel: %s (retrying later)", this);
293
294 GravityConfig gravityConfig = getGravity().getGravityConfig();
295 if (gravityConfig.isRetryOnError()) {
296 receivedQueueLock.lock();
297 try {
298 if (receivedQueue.size() + messages.size() > gravityConfig.getMaxMessagesQueuedPerChannel()) {
299 log.warn(
300 "Channel %s has reached its maximum queue capacity %s (throwing %s messages)",
301 this,
302 gravityConfig.getMaxMessagesQueuedPerChannel(),
303 messages.size()
304 );
305 }
306 else
307 receivedQueue.addAll(0, messages);
308 }
309 finally {
310 receivedQueueLock.unlock();
311 }
312 }
313
314 return true; // Messages weren't delivered, but http context isn't valid anymore.
315 }
316 finally {
317 // Cleanup serialization context (thread local)
318 try {
319 GraniteContext.release();
320 }
321 catch (Exception e) {
322 // should never happen...
323 }
324
325 // Close output stream.
326 try {
327 if (os != null) {
328 try {
329 os.close();
330 }
331 catch (IOException e) {
332 log.warn(e, "Could not close output stream (ignored)");
333 }
334 }
335 }
336 finally {
337 // Cleanup http context (only if this method wasn't explicitly called with a non null
338 // AsyncHttpContext from the servlet).
339 if (!httpAsParam)
340 releaseAsyncHttpContext(asyncHttpContext);
341 }
342 }
343 }
344
345 public void destroy() {
346 Gravity gravity = getGravity();
347 gravity.cancel(publisher);
348 gravity.cancel(receiver);
349
350 subscriptions.clear();
351 }
352
353 ///////////////////////////////////////////////////////////////////////////
354 // Protected utilities.
355
356 protected boolean queueReceiver() {
357 if (hasReceivedMessage()) {
358 receiver.queue(getGravity());
359 return true;
360 }
361 return false;
362 }
363
364 ///////////////////////////////////////////////////////////////////////////
365 // Object overwritten methods.
366
367 @Override
368 public boolean equals(Object obj) {
369 return (obj instanceof Channel && id.equals(((Channel)obj).getId()));
370 }
371
372 @Override
373 public int hashCode() {
374 return id.hashCode();
375 }
376
377 @Override
378 public String toString() {
379 return getClass().getName() + " {id=" + id + ", subscriptions=" + subscriptions.values() + "}";
380 }
381 }