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.channel.amf;
022
023 import java.io.IOException;
024 import java.io.InputStream;
025 import java.io.UnsupportedEncodingException;
026 import java.net.URI;
027 import java.util.Map;
028 import java.util.TimerTask;
029 import java.util.concurrent.ConcurrentHashMap;
030 import java.util.concurrent.ConcurrentMap;
031 import java.util.concurrent.TimeUnit;
032 import java.util.concurrent.atomic.AtomicReference;
033
034 import org.granite.client.configuration.Configuration;
035 import org.granite.client.configuration.DefaultConfiguration;
036 import org.granite.client.messaging.Consumer;
037 import org.granite.client.messaging.ResponseListener;
038 import org.granite.client.messaging.channel.AsyncToken;
039 import org.granite.client.messaging.channel.Channel;
040 import org.granite.client.messaging.channel.MessagingChannel;
041 import org.granite.client.messaging.channel.ResponseMessageFuture;
042 import org.granite.client.messaging.codec.AMF3MessagingCodec;
043 import org.granite.client.messaging.codec.MessagingCodec;
044 import org.granite.client.messaging.messages.RequestMessage;
045 import org.granite.client.messaging.messages.ResponseMessage;
046 import org.granite.client.messaging.messages.requests.DisconnectMessage;
047 import org.granite.client.messaging.messages.responses.AbstractResponseMessage;
048 import org.granite.client.messaging.messages.responses.ResultMessage;
049 import org.granite.client.messaging.transport.DefaultTransportMessage;
050 import org.granite.client.messaging.transport.Transport;
051 import org.granite.client.messaging.transport.TransportMessage;
052 import org.granite.logging.Logger;
053 import org.granite.util.UUIDUtil;
054
055 import flex.messaging.messages.AcknowledgeMessage;
056 import flex.messaging.messages.AsyncMessage;
057 import flex.messaging.messages.CommandMessage;
058 import flex.messaging.messages.Message;
059
060 /**
061 * @author Franck WOLFF
062 */
063 public class AMFMessagingChannel extends AbstractAMFChannel implements MessagingChannel {
064
065 private static final Logger log = Logger.getLogger(AMFMessagingChannel.class);
066
067 protected final MessagingCodec<Message[]> codec;
068
069 protected String sessionId = null;
070 protected final ConcurrentMap<String, Consumer> consumersMap = new ConcurrentHashMap<String, Consumer>();
071 protected final AtomicReference<String> connectMessageId = new AtomicReference<String>(null);
072 protected final AtomicReference<ReconnectTimerTask> reconnectTimerTask = new AtomicReference<ReconnectTimerTask>();
073
074 protected volatile long reconnectIntervalMillis = TimeUnit.SECONDS.toMillis(30L);
075 protected volatile long reconnectMaxAttempts = 60L;
076 protected volatile long reconnectAttempts = 0L;
077
078 public AMFMessagingChannel(Transport transport, String id, URI uri) {
079 this(transport, DefaultConfiguration.getInstance(), id, uri);
080 }
081
082 public AMFMessagingChannel(Transport transport, Configuration configuration, String id, URI uri) {
083 super(transport, id, uri, 1);
084
085 this.codec = newMessagingCodec(configuration);
086 }
087
088 protected MessagingCodec<Message[]> newMessagingCodec(Configuration configuration) {
089 return new AMF3MessagingCodec(configuration);
090 }
091
092 public void setSessionId(String sessionId) {
093 if ((sessionId == null && this.sessionId != null) || !sessionId.equals(this.sessionId)) {
094 this.sessionId = sessionId;
095 log.info("Received sessionId %s", sessionId);
096 }
097 }
098
099 protected boolean connect() {
100
101 // Connecting: make sure we don't have an active reconnect timer task.
102 cancelReconnectTimerTask();
103
104 // No subscriptions...
105 if (consumersMap.isEmpty())
106 return false;
107
108 // We are already waiting for a connection/answer.
109 final String id = UUIDUtil.randomUUID();
110 if (!connectMessageId.compareAndSet(null, id))
111 return false;
112
113 log.debug("Connecting channel with clientId %s", clientId);
114
115 // Create and try to send the connect message.
116 CommandMessage connectMessage = new CommandMessage();
117 connectMessage.setOperation(CommandMessage.CONNECT_OPERATION);
118 connectMessage.setMessageId(id);
119 connectMessage.setTimestamp(System.currentTimeMillis());
120 connectMessage.setClientId(clientId);
121
122 try {
123 transport.send(this, new DefaultTransportMessage<Message[]>(id, true, clientId, sessionId, new Message[]{connectMessage}, codec));
124
125 return true;
126 }
127 catch (Exception e) {
128 // Connect immediately failed, release the message id and schedule a reconnect.
129 connectMessageId.set(null);
130 scheduleReconnectTimerTask();
131
132 return false;
133 }
134 }
135
136 @Override
137 public void addConsumer(Consumer consumer) {
138 consumersMap.putIfAbsent(consumer.getSubscriptionId(), consumer);
139
140 connect();
141 }
142
143 @Override
144 public boolean removeConsumer(Consumer consumer) {
145 return (consumersMap.remove(consumer.getSubscriptionId()) != null);
146 }
147
148 public synchronized ResponseMessageFuture disconnect(ResponseListener...listeners) {
149 cancelReconnectTimerTask();
150
151 connectMessageId.set(null);
152 reconnectAttempts = 0L;
153
154 for (Consumer consumer : consumersMap.values())
155 consumer.onDisconnect();
156
157 consumersMap.clear();
158
159 return send(new DisconnectMessage(clientId), listeners);
160 }
161
162 @Override
163 protected TransportMessage createTransportMessage(AsyncToken token) throws UnsupportedEncodingException {
164 Message[] messages = convertToAmf(token.getRequest());
165 return new DefaultTransportMessage<Message[]>(token.getId(), false, clientId, sessionId, messages, codec);
166 }
167
168 @Override
169 protected ResponseMessage decodeResponse(InputStream is) throws IOException {
170 boolean reconnect = true;
171
172 try {
173 if (is.available() > 0) {
174 final Message[] messages = codec.decode(is);
175
176 if (messages.length > 0 && messages[0] instanceof AcknowledgeMessage) {
177
178 reconnect = false;
179
180 final AbstractResponseMessage response = convertFromAmf((AcknowledgeMessage)messages[0]);
181
182 if (response instanceof ResultMessage) {
183 RequestMessage request = getRequest(response.getCorrelationId());
184 if (request != null) {
185 ResultMessage result = (ResultMessage)response;
186 switch (request.getType()) {
187
188 case PING:
189 if (messages[0].getBody() instanceof Map) {
190 Map<?, ?> advices = (Map<?, ?>)messages[0].getBody();
191 Object reconnectIntervalMillis = advices.get(Channel.RECONNECT_INTERVAL_MS_KEY);
192 if (reconnectIntervalMillis instanceof Number)
193 this.reconnectIntervalMillis = ((Number)reconnectIntervalMillis).longValue();
194 Object reconnectMaxAttempts = advices.get(Channel.RECONNECT_MAX_ATTEMPTS_KEY);
195 if (reconnectMaxAttempts instanceof Number)
196 this.reconnectMaxAttempts = ((Number)reconnectMaxAttempts).longValue();
197 }
198 break;
199
200 case SUBSCRIBE:
201 result.setResult(messages[0].getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER));
202 break;
203
204 default:
205 break;
206 }
207 }
208 }
209
210 AbstractResponseMessage current = response;
211 for (int i = 1; i < messages.length; i++) {
212 if (!(messages[i] instanceof AcknowledgeMessage))
213 throw new RuntimeException("Message should be an AcknowledgeMessage: " + messages[i]);
214
215 AbstractResponseMessage next = convertFromAmf((AcknowledgeMessage)messages[i]);
216 current.setNext(next);
217 current = next;
218 }
219
220 return response;
221 }
222
223 for (Message message : messages) {
224 if (!(message instanceof AsyncMessage))
225 throw new RuntimeException("Message should be an AsyncMessage: " + message);
226
227 String subscriptionId = (String)message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER);
228 Consumer consumer = consumersMap.get(subscriptionId);
229 if (consumer != null)
230 consumer.onMessage(convertFromAmf((AsyncMessage)message));
231 else
232 log.warn("No consumer for subscriptionId: %s", subscriptionId);
233 }
234 }
235 }
236 finally {
237 if (reconnect) {
238 connectMessageId.set(null);
239 connect();
240 }
241 }
242
243 return null;
244 }
245
246 @Override
247 public void onError(TransportMessage message, Exception e) {
248 super.onError(message, e);
249
250 if (message != null && connectMessageId.compareAndSet(message.getId(), null))
251 scheduleReconnectTimerTask();
252 }
253
254 protected void cancelReconnectTimerTask() {
255 ReconnectTimerTask task = reconnectTimerTask.getAndSet(null);
256 if (task != null && task.cancel())
257 reconnectAttempts = 0L;
258 }
259
260 protected void scheduleReconnectTimerTask() {
261 ReconnectTimerTask task = new ReconnectTimerTask();
262
263 ReconnectTimerTask previousTask = reconnectTimerTask.getAndSet(task);
264 if (previousTask != null)
265 previousTask.cancel();
266
267 if (reconnectAttempts < reconnectMaxAttempts) {
268 reconnectAttempts++;
269 schedule(task, reconnectIntervalMillis);
270 }
271 }
272
273 class ReconnectTimerTask extends TimerTask {
274
275 @Override
276 public void run() {
277 connect();
278 }
279 }
280 }