001 package org.granite.gravity.jetty8;
002
003 import java.io.ByteArrayInputStream;
004 import java.io.ByteArrayOutputStream;
005 import java.io.IOException;
006 import java.io.ObjectInput;
007 import java.io.ObjectOutput;
008 import java.util.Arrays;
009 import java.util.LinkedList;
010
011 import javax.servlet.ServletContext;
012
013 import org.eclipse.jetty.websocket.WebSocket;
014 import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage;
015 import org.granite.context.GraniteContext;
016 import org.granite.gravity.AbstractChannel;
017 import org.granite.gravity.AsyncHttpContext;
018 import org.granite.gravity.Gravity;
019 import org.granite.gravity.GravityConfig;
020 import org.granite.logging.Logger;
021 import org.granite.messaging.webapp.ServletGraniteContext;
022
023 import flex.messaging.messages.AsyncMessage;
024 import flex.messaging.messages.Message;
025
026
027 public class JettyWebSocketChannel extends AbstractChannel implements WebSocket, OnBinaryMessage {
028
029 private static final Logger log = Logger.getLogger(JettyWebSocketChannel.class);
030
031 private ServletContext servletContext;
032 private Connection connection;
033 private Message connectAckMessage;
034
035
036 public JettyWebSocketChannel(Gravity gravity, String id, JettyWebSocketChannelFactory factory, ServletContext servletContext, String clientType) {
037 super(gravity, id, factory, clientType);
038 this.servletContext = servletContext;
039 }
040
041 public void setConnectAckMessage(Message ackMessage) {
042 this.connectAckMessage = ackMessage;
043 }
044
045 public void onOpen(Connection connection) {
046 this.connection = connection;
047 this.connection.setMaxIdleTime((int)getGravity().getGravityConfig().getChannelIdleTimeoutMillis());
048
049 log.debug("WebSocket connection onOpen");
050
051 if (connectAckMessage == null)
052 return;
053
054 try {
055 initializeRequest();
056
057 // Return an acknowledge message with the server-generated clientId
058 byte[] resultData = serialize(getGravity(), new Message[] { connectAckMessage });
059
060 connection.sendMessage(resultData, 0, resultData.length);
061
062 connectAckMessage = null;
063 }
064 catch (IOException e) {
065 log.error(e, "Could not send connect acknowledge");
066 }
067 finally {
068 cleanupRequest();
069 }
070 }
071
072 public void onClose(int closeCode, String message) {
073 log.debug("WebSocket connection onClose %d, %s", closeCode, message);
074 }
075
076 public void onMessage(byte[] data, int offset, int length) {
077 log.debug("WebSocket connection onMessage %d", data.length);
078
079 try {
080 initializeRequest();
081
082 Message[] messages = deserialize(getGravity(), data, offset, length);
083
084 log.debug(">> [AMF3 REQUESTS] %s", (Object)messages);
085
086 Message[] responses = null;
087
088 boolean accessed = false;
089 int responseIndex = 0;
090 for (int i = 0; i < messages.length; i++) {
091 Message message = messages[i];
092
093 // Ask gravity to create a specific response (will be null with a connect request from tunnel).
094 Message response = getGravity().handleMessage(getFactory(), message);
095 String channelId = (String)message.getClientId();
096
097 // Mark current channel (if any) as accessed.
098 if (!accessed)
099 accessed = getGravity().access(channelId);
100
101 if (response != null) {
102 if (responses == null)
103 responses = new Message[1];
104 else
105 responses = Arrays.copyOf(responses, responses.length+1);
106 responses[responseIndex++] = response;
107 }
108 }
109
110 if (responses != null && responses.length > 0) {
111 log.debug("<< [AMF3 RESPONSES] %s", (Object)responses);
112
113 byte[] resultData = serialize(getGravity(), responses);
114
115 connection.sendMessage(resultData, 0, resultData.length);
116 }
117 }
118 catch (ClassNotFoundException e) {
119 log.error(e, "Could not handle incoming message data");
120 }
121 catch (IOException e) {
122 log.error(e, "Could not handle incoming message data");
123 }
124 finally {
125 cleanupRequest();
126 }
127 }
128
129 private Gravity initializeRequest() {
130 ServletGraniteContext.createThreadInstance(gravity.getGraniteConfig(), gravity.getServicesConfig(), servletContext, sessionId, clientType);
131 return gravity;
132 }
133
134 private static Message[] deserialize(Gravity gravity, byte[] data, int offset, int length) throws ClassNotFoundException, IOException {
135 ByteArrayInputStream is = new ByteArrayInputStream(data, offset, length);
136 try {
137 ObjectInput amf3Deserializer = gravity.getGraniteConfig().newAMF3Deserializer(is);
138 Object[] objects = (Object[])amf3Deserializer.readObject();
139 Message[] messages = new Message[objects.length];
140 System.arraycopy(objects, 0, messages, 0, objects.length);
141
142 return messages;
143 }
144 finally {
145 is.close();
146 }
147 }
148
149 private static byte[] serialize(Gravity gravity, Message[] messages) throws IOException {
150 ByteArrayOutputStream os = null;
151 try {
152 os = new ByteArrayOutputStream(200*messages.length);
153 ObjectOutput amf3Serializer = gravity.getGraniteConfig().newAMF3Serializer(os);
154 amf3Serializer.writeObject(messages);
155 os.flush();
156 return os.toByteArray();
157 }
158 finally {
159 if (os != null)
160 os.close();
161 }
162 }
163
164 private static void cleanupRequest() {
165 GraniteContext.release();
166 }
167
168 @Override
169 public boolean runReceived(AsyncHttpContext asyncHttpContext) {
170
171 LinkedList<AsyncMessage> messages = null;
172 ByteArrayOutputStream os = null;
173
174 try {
175 receivedQueueLock.lock();
176 try {
177 // Do we have any pending messages?
178 if (receivedQueue.isEmpty())
179 return false;
180
181 // Both conditions are ok, get all pending messages.
182 messages = receivedQueue;
183 receivedQueue = new LinkedList<AsyncMessage>();
184 }
185 finally {
186 receivedQueueLock.unlock();
187 }
188
189 if (connection == null || !connection.isOpen())
190 return false;
191
192 AsyncMessage[] messagesArray = new AsyncMessage[messages.size()];
193 int i = 0;
194 for (AsyncMessage message : messages)
195 messagesArray[i++] = message;
196
197 // Setup serialization context (thread local)
198 Gravity gravity = getGravity();
199 GraniteContext context = ServletGraniteContext.createThreadInstance(gravity.getGraniteConfig(), gravity.getServicesConfig(), servletContext, sessionId, clientType);
200
201 os = new ByteArrayOutputStream(500);
202 ObjectOutput amf3Serializer = context.getGraniteConfig().newAMF3Serializer(os);
203
204 log.debug("<< [MESSAGES for channel=%s] %s", this, messagesArray);
205
206 amf3Serializer.writeObject(messagesArray);
207
208 connection.sendMessage(os.toByteArray(), 0, os.size());
209
210 return true; // Messages were delivered
211 }
212 catch (IOException e) {
213 log.warn(e, "Could not send messages to channel: %s (retrying later)", this);
214
215 GravityConfig gravityConfig = getGravity().getGravityConfig();
216 if (gravityConfig.isRetryOnError()) {
217 receivedQueueLock.lock();
218 try {
219 if (receivedQueue.size() + messages.size() > gravityConfig.getMaxMessagesQueuedPerChannel()) {
220 log.warn(
221 "Channel %s has reached its maximum queue capacity %s (throwing %s messages)",
222 this,
223 gravityConfig.getMaxMessagesQueuedPerChannel(),
224 messages.size()
225 );
226 }
227 else
228 receivedQueue.addAll(0, messages);
229 }
230 finally {
231 receivedQueueLock.unlock();
232 }
233 }
234
235 return true; // Messages weren't delivered, but http context isn't valid anymore.
236 }
237 finally {
238 if (os != null) {
239 try {
240 os.close();
241 }
242 catch (Exception e) {
243 // Could not close bytearray ???
244 }
245 }
246
247 // Cleanup serialization context (thread local)
248 try {
249 GraniteContext.release();
250 }
251 catch (Exception e) {
252 // should never happen...
253 }
254 }
255 }
256
257 @Override
258 public void destroy() {
259 try {
260 super.destroy();
261 }
262 finally {
263 close();
264 }
265 }
266
267 public void close() {
268 if (connection != null) {
269 connection.close(1000, "Channel closed");
270 connection = null;
271 }
272 }
273
274 @Override
275 protected boolean hasAsyncHttpContext() {
276 return true;
277 }
278
279 @Override
280 protected void releaseAsyncHttpContext(AsyncHttpContext context) {
281 }
282
283 @Override
284 protected AsyncHttpContext acquireAsyncHttpContext() {
285 return null;
286 }
287 }