001 /*
002 GRANITE DATA SERVICES
003 Copyright (C) 2012 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.client.messaging.transport.jetty;
022
023 import java.io.ByteArrayInputStream;
024 import java.io.IOException;
025 import java.net.URI;
026 import java.util.LinkedList;
027 import java.util.concurrent.Future;
028 import java.util.concurrent.TimeoutException;
029
030 import org.eclipse.jetty.websocket.WebSocket.Connection;
031 import org.eclipse.jetty.websocket.WebSocket.OnBinaryMessage;
032 import org.eclipse.jetty.websocket.WebSocketClient;
033 import org.eclipse.jetty.websocket.WebSocketClientFactory;
034 import org.granite.client.messaging.channel.Channel;
035 import org.granite.client.messaging.transport.AbstractTransport;
036 import org.granite.client.messaging.transport.TransportException;
037 import org.granite.client.messaging.transport.TransportFuture;
038 import org.granite.client.messaging.transport.TransportMessage;
039 import org.granite.client.messaging.transport.WebSocketTransport;
040 import org.granite.logging.Logger;
041 import org.granite.util.PublicByteArrayOutputStream;
042
043
044 /**
045 * @author William DRAI
046 */
047 public class JettyWebSocketTransport extends AbstractTransport implements WebSocketTransport {
048
049 private static final Logger log = Logger.getLogger(JettyWebSocketTransport.class);
050
051 private final static int CLOSE_NORMAL = 1000;
052 private final static int CLOSE_SHUTDOWN = 1001;
053 // private final static int CLOSE_PROTOCOL = 1002;
054
055 private WebSocketClientFactory webSocketClientFactory = null;
056
057 private Future<Connection> connectionFuture = null;
058 private boolean connected = false;
059
060 private int maxIdleTime = 3000000;
061 private int reconnectMaxAttempts = 5;
062 private int reconnectIntervalMillis = 60000;
063
064 public void setMaxIdleTime(int maxIdleTime) {
065 this.maxIdleTime = maxIdleTime;
066 }
067
068 @Override
069 public synchronized boolean start() {
070 if (webSocketClientFactory != null && webSocketClientFactory.isStarted())
071 return true;
072
073 stop();
074
075 log.info("Starting Jetty WebSocketClient transport...");
076
077 try {
078 webSocketClientFactory = new WebSocketClientFactory();
079 webSocketClientFactory.setBufferSize(4096);
080 webSocketClientFactory.start();
081
082 final long timeout = System.currentTimeMillis() + 10000L; // 10sec.
083 while (!webSocketClientFactory.isStarted()) {
084 if (System.currentTimeMillis() > timeout)
085 throw new TimeoutException("Jetty WebSocketFactory start process too long");
086 Thread.sleep(100);
087 }
088
089 log.info("Jetty WebSocketClient transport started.");
090 return true;
091 }
092 catch (Exception e) {
093 webSocketClientFactory = null;
094 getStatusHandler().handleException(new TransportException("Could not start Jetty WebSocketFactory", e));
095
096 log.error(e, "Jetty WebSocketClient transport failed to start.");
097 return false;
098 }
099 }
100
101 public boolean isStarted() {
102 return webSocketClientFactory != null && webSocketClientFactory.isStarted();
103 }
104
105 @Override
106 public TransportFuture send(final Channel channel, final TransportMessage message) {
107
108 synchronized (channel) {
109
110 TransportData transportData = channel.getTransportData();
111 if (transportData == null) {
112 transportData = new TransportData();
113 channel.setTransportData(transportData);
114 }
115
116 if (message != null) {
117 if (message.isConnect())
118 connectMessage = message;
119 else
120 transportData.pendingMessages.addLast(message);
121 }
122
123 if (transportData.connection == null) {
124 connect(channel, message);
125 return null;
126 }
127
128 while (!transportData.pendingMessages.isEmpty()) {
129 TransportMessage pendingMessage = transportData.pendingMessages.removeFirst();
130 try {
131 PublicByteArrayOutputStream os = new PublicByteArrayOutputStream(256);
132 pendingMessage.encode(os);
133 byte[] data = os.getBytes();
134 transportData.connection.sendMessage(data, 0, os.size());
135 }
136 catch (IOException e) {
137 transportData.pendingMessages.addFirst(pendingMessage);
138 // report error...
139 break;
140 }
141 }
142 }
143
144 return null;
145 }
146
147 @Override
148 public void poll(final Channel channel, final TransportMessage message) {
149 send(channel, message);
150 }
151
152 private int reconnectAttempts = 0;
153 private TransportMessage connectMessage = null;
154
155 public Future<Connection> connect(final Channel channel, final TransportMessage transportMessage) {
156 if (connectionFuture != null)
157 return connectionFuture;
158
159 connected = true;
160
161 URI uri = channel.getUri();
162
163 try {
164 WebSocketClient webSocketClient = webSocketClientFactory.newWebSocketClient();
165 webSocketClient.setMaxIdleTime(maxIdleTime);
166 webSocketClient.setMaxTextMessageSize(1024);
167 webSocketClient.setProtocol("org.granite.gravity");
168
169 if (transportMessage.getSessionId() != null)
170 webSocketClient.getCookies().put("JSESSIONID", transportMessage.getSessionId());
171
172 String u = uri.toString();
173 u += "?connectId=" + transportMessage.getId() + "&GDSClientType=java";
174 if (transportMessage.getClientId() != null)
175 u += "&GDSClientId=" + transportMessage.getClientId();
176 else if (channel.getClientId() != null)
177 u += "&GDSClientId=" + channel.getClientId();
178
179 connectionFuture = webSocketClient.open(new URI(u), new OnBinaryMessage() {
180
181 @Override
182 public void onOpen(Connection connection) {
183 synchronized (channel) {
184 connectionFuture = null;
185 reconnectAttempts = 0;
186 ((TransportData)channel.getTransportData()).connection = connection;
187 send(channel, null);
188 }
189 }
190
191 @Override
192 public void onMessage(byte[] data, int offset, int length) {
193 channel.onMessage(new ByteArrayInputStream(data, offset, length));
194 }
195
196 @Override
197 public void onClose(int closeCode, String message) {
198 boolean waitBeforeReconnect = !(closeCode == CLOSE_NORMAL && message.startsWith("Idle"));
199
200 synchronized (channel) {
201 // Mark the connection as close, the channel should reopen a connection for the next message
202 ((TransportData)channel.getTransportData()).connection = null;
203 connectionFuture = null;
204
205 if (!isStarted())
206 connected = false;
207
208 if (closeCode == CLOSE_SHUTDOWN) {
209 connected = false;
210 return;
211 }
212
213 if (channel.getClientId() == null) {
214 getStatusHandler().handleException(new TransportException("Transport could not connect code: " + closeCode + " " + message));
215 return;
216 }
217
218 if (connected) {
219 if (reconnectAttempts >= reconnectMaxAttempts) {
220 connected = false;
221 if (isStarted())
222 stop();
223
224 channel.onError(transportMessage, new RuntimeException(message + " (code=" + closeCode + ")"));
225 getStatusHandler().handleException(new TransportException("Transport disconnected"));
226 return;
227 }
228
229 if (waitBeforeReconnect) {
230 try {
231 waitBeforeReconnect = false;
232 Thread.sleep(reconnectIntervalMillis);
233 }
234 catch (InterruptedException e) {
235 }
236 }
237
238 reconnectAttempts++;
239
240 // If the channel should be connected, try to reconnect
241 log.info("Connection lost (code %d, msg %s), reconnect channel (retry #%d)", closeCode, message, reconnectAttempts);
242 connect(channel, connectMessage);
243 }
244 }
245 }
246 });
247
248 return connectionFuture;
249 }
250 catch (Exception e) {
251 getStatusHandler().handleException(new TransportException("Could not connect to uri " + channel.getUri(), e));
252
253 return null;
254 }
255 }
256
257 private static class TransportData {
258
259 private final LinkedList<TransportMessage> pendingMessages = new LinkedList<TransportMessage>();
260 private Connection connection = null;
261 }
262
263 @Override
264 public synchronized void stop() {
265 if (webSocketClientFactory == null)
266 return;
267
268 log.info("Stopping Jetty WebSocketClient transport...");
269
270 super.stop();
271
272 try {
273 webSocketClientFactory.stop();
274 }
275 catch (Exception e) {
276 getStatusHandler().handleException(new TransportException("Could not stop Jetty WebSocketFactory", e));
277
278 log.error(e, "Jetty WebSocketClient failed to stop properly.");
279 }
280 finally {
281 webSocketClientFactory = null;
282 }
283
284 log.info("Jetty WebSocketClient transport stopped.");
285 }
286 }