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