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.UnsupportedEncodingException;
024 import java.net.URI;
025 import java.util.ArrayList;
026 import java.util.List;
027
028 import org.granite.client.messaging.channel.AbstractHTTPChannel;
029 import org.granite.client.messaging.channel.Credentials;
030 import org.granite.client.messaging.channel.UsernamePasswordCredentials;
031 import org.granite.client.messaging.messages.RequestMessage;
032 import org.granite.client.messaging.messages.push.TopicMessage;
033 import org.granite.client.messaging.messages.requests.InvocationMessage;
034 import org.granite.client.messaging.messages.requests.LoginMessage;
035 import org.granite.client.messaging.messages.requests.PublishMessage;
036 import org.granite.client.messaging.messages.requests.SubscribeMessage;
037 import org.granite.client.messaging.messages.requests.UnsubscribeMessage;
038 import org.granite.client.messaging.messages.responses.AbstractResponseMessage;
039 import org.granite.client.messaging.messages.responses.FaultMessage;
040 import org.granite.client.messaging.messages.responses.FaultMessage.Code;
041 import org.granite.client.messaging.messages.responses.ResultMessage;
042 import org.granite.client.messaging.transport.Transport;
043 import org.granite.messaging.service.security.SecurityServiceException;
044
045 import flex.messaging.messages.AcknowledgeMessage;
046 import flex.messaging.messages.AsyncMessage;
047 import flex.messaging.messages.CommandMessage;
048 import flex.messaging.messages.ErrorMessage;
049 import flex.messaging.messages.Message;
050 import flex.messaging.messages.RemotingMessage;
051
052 /**
053 * @author Franck WOLFF
054 */
055 public abstract class AbstractAMFChannel extends AbstractHTTPChannel {
056
057 public AbstractAMFChannel(Transport transport, String id, URI uri) {
058 super(transport, id, uri);
059 }
060
061 public AbstractAMFChannel(Transport transport, String id, URI uri, int maxConcurrentRequests) {
062 super(transport, id, uri, maxConcurrentRequests);
063 }
064
065 protected Message[] convertToAmf(RequestMessage request) throws UnsupportedEncodingException {
066 Message[] messages = null;
067
068 switch (request.getType()) {
069 case PING: {
070 CommandMessage commandMessage = new CommandMessage();
071 commandMessage.setOperation(CommandMessage.CLIENT_PING_OPERATION);
072 commandMessage.setMessageId(request.getId());
073 commandMessage.setTimestamp(request.getTimestamp());
074 commandMessage.setTimeToLive(request.getTimeToLive());
075 commandMessage.setHeaders(request.getHeaders());
076 messages = new Message[]{commandMessage};
077 break;
078 }
079 case LOGIN: {
080 LoginMessage login = (LoginMessage)request;
081 CommandMessage commandMessage = new CommandMessage();
082 commandMessage.setOperation(CommandMessage.LOGIN_OPERATION);
083 commandMessage.setMessageId(request.getId());
084 commandMessage.setTimestamp(request.getTimestamp());
085 commandMessage.setTimeToLive(request.getTimeToLive());
086 commandMessage.setHeaders(request.getHeaders());
087 Credentials credentials = login.getCredentials();
088 if (credentials instanceof UsernamePasswordCredentials)
089 commandMessage.setBody(((UsernamePasswordCredentials)credentials).encodeBase64());
090 else
091 throw new UnsupportedOperationException("Unsupported credentials type: " + credentials);
092 messages = new Message[]{commandMessage};
093 break;
094 }
095 case LOGOUT: {
096 CommandMessage commandMessage = new CommandMessage();
097 commandMessage.setOperation(CommandMessage.LOGOUT_OPERATION);
098 commandMessage.setMessageId(request.getId());
099 commandMessage.setTimestamp(request.getTimestamp());
100 commandMessage.setTimeToLive(request.getTimeToLive());
101 commandMessage.setHeaders(request.getHeaders());
102 messages = new Message[]{commandMessage};
103 break;
104 }
105 case PUBLISH: {
106 PublishMessage publish = (PublishMessage)request;
107 AsyncMessage asyncMessage = new AsyncMessage();
108 asyncMessage.setMessageId(publish.getId());
109 asyncMessage.setTimestamp(publish.getTimestamp());
110 asyncMessage.setTimeToLive(publish.getTimeToLive());
111 asyncMessage.setHeaders(publish.getHeaders());
112 asyncMessage.setDestination(publish.getDestination());
113 asyncMessage.setHeader(AsyncMessage.SUBTOPIC_HEADER, publish.getTopic());
114 asyncMessage.setBody(publish.getBody());
115 messages = new Message[]{asyncMessage};
116 break;
117 }
118 case SUBSCRIBE: {
119 SubscribeMessage subscribe = (SubscribeMessage)request;
120 CommandMessage commandMessage = new CommandMessage();
121 commandMessage.setOperation(CommandMessage.SUBSCRIBE_OPERATION);
122 commandMessage.setMessageId(subscribe.getId());
123 commandMessage.setTimestamp(subscribe.getTimestamp());
124 commandMessage.setTimeToLive(subscribe.getTimeToLive());
125 commandMessage.setHeaders(subscribe.getHeaders());
126 commandMessage.setDestination(subscribe.getDestination());
127 commandMessage.setHeader(AsyncMessage.SUBTOPIC_HEADER, subscribe.getTopic());
128 if (subscribe.getSelector() != null)
129 commandMessage.setHeader(CommandMessage.SELECTOR_HEADER, subscribe.getSelector());
130 messages = new Message[]{commandMessage};
131 break;
132 }
133 case UNSUBSCRIBE: {
134 UnsubscribeMessage unsubscribe = (UnsubscribeMessage)request;
135 CommandMessage commandMessage = new CommandMessage();
136 commandMessage.setOperation(CommandMessage.UNSUBSCRIBE_OPERATION);
137 commandMessage.setMessageId(unsubscribe.getId());
138 commandMessage.setTimestamp(unsubscribe.getTimestamp());
139 commandMessage.setTimeToLive(unsubscribe.getTimeToLive());
140 commandMessage.setHeaders(unsubscribe.getHeaders());
141 commandMessage.setDestination(unsubscribe.getDestination());
142 commandMessage.setHeader(AsyncMessage.SUBTOPIC_HEADER, unsubscribe.getTopic());
143 commandMessage.setHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER, unsubscribe.getSubscriptionId());
144 messages = new Message[]{commandMessage};
145 break;
146 }
147 case INVOCATION: {
148 List<Message> remotingMessages = new ArrayList<Message>();
149
150 for (InvocationMessage invocation : (InvocationMessage)request) {
151 RemotingMessage remotingMessage = new RemotingMessage();
152 remotingMessage.setDestination(invocation.getServiceId());
153 remotingMessage.setOperation(invocation.getMethod());
154 remotingMessage.setBody(invocation.getParameters() != null ? invocation.getParameters() : new Object[0]);
155 remotingMessage.setMessageId(invocation.getId());
156 remotingMessage.setTimestamp(invocation.getTimestamp());
157 remotingMessage.setTimeToLive(invocation.getTimeToLive());
158 remotingMessage.setHeaders(invocation.getHeaders());
159 remotingMessages.add(remotingMessage);
160 }
161
162 messages = remotingMessages.toArray(new Message[remotingMessages.size()]);
163 break;
164 }
165 case DISCONNECT: {
166 CommandMessage commandMessage = new CommandMessage();
167 commandMessage.setOperation(CommandMessage.DISCONNECT_OPERATION);
168 commandMessage.setMessageId(request.getId());
169 commandMessage.setTimestamp(request.getTimestamp());
170 commandMessage.setTimeToLive(request.getTimeToLive());
171 commandMessage.setHeaders(request.getHeaders());
172 messages = new Message[]{commandMessage};
173 break;
174 }
175 default:
176 throw new IllegalArgumentException("Unsupported message type: " + request);
177 }
178
179 for (Message message : messages) {
180 message.setClientId(getClientId());
181 message.setHeader(Message.ENDPOINT_HEADER, getId());
182 }
183
184 return messages;
185 }
186
187 protected TopicMessage convertFromAmf(AsyncMessage message) {
188 return new TopicMessage(
189 message.getMessageId(),
190 (String)message.getClientId(),
191 message.getTimestamp(),
192 message.getTimeToLive(),
193 message.getHeaders(),
194 message.getBody()
195 );
196 }
197
198 protected AbstractResponseMessage convertFromAmf(AcknowledgeMessage message) {
199 if (message instanceof ErrorMessage) {
200 ErrorMessage errorMessage = (ErrorMessage)message;
201
202 Code code = Code.UNKNOWN;
203
204 String flexCode = errorMessage.getFaultCode();
205 if (ErrorMessage.CODE_SERVER_CALL_FAILED.equals(flexCode))
206 code = Code.SERVER_CALL_FAILED;
207 else if (SecurityServiceException.CODE_ACCESS_DENIED.equals(flexCode))
208 code = Code.ACCESS_DENIED;
209 else if (SecurityServiceException.CODE_INVALID_CREDENTIALS.equals(flexCode)) {
210 authenticated = false;
211 code = Code.INVALID_CREDENTIALS;
212 }
213 else if (SecurityServiceException.CODE_AUTHENTICATION_FAILED.equals(flexCode)) {
214 authenticated = false;
215 code = Code.AUTHENTICATION_FAILED;
216 }
217 else if (SecurityServiceException.CODE_NOT_LOGGED_IN.equals(flexCode)) {
218 authenticated = false;
219 code = Code.NOT_LOGGED_IN;
220 }
221 else if (SecurityServiceException.CODE_SESSION_EXPIRED.equals(flexCode)) {
222 authenticated = false;
223 code = Code.SESSION_EXPIRED;
224 }
225 else if ("Validation.Failed".equals(flexCode))
226 code = Code.VALIDATION_FAILED;
227 else if ("Persistence.OptimisticLock".equals(flexCode))
228 code = Code.OPTIMISTIC_LOCK;
229
230 FaultMessage fault = new FaultMessage(
231 errorMessage.getMessageId(),
232 (String)errorMessage.getClientId(),
233 errorMessage.getTimestamp(),
234 errorMessage.getTimeToLive(),
235 errorMessage.getHeaders(),
236 errorMessage.getCorrelationId(),
237 code,
238 errorMessage.getFaultString(),
239 errorMessage.getFaultDetail(),
240 errorMessage.getRootCause(),
241 errorMessage.getExtendedData()
242 );
243
244 if (code == Code.UNKNOWN)
245 fault.setUnknownCode(flexCode);
246
247 return fault;
248 }
249
250 return new ResultMessage(
251 message.getMessageId(),
252 (String)message.getClientId(),
253 message.getTimestamp(),
254 message.getTimeToLive(),
255 message.getHeaders(),
256 message.getCorrelationId(),
257 message.getBody()
258 );
259 }
260 }