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.apache;
022
023 import java.io.IOException;
024 import java.io.InputStream;
025 import java.util.concurrent.Future;
026 import java.util.concurrent.TimeoutException;
027
028 import org.apache.http.HttpResponse;
029 import org.apache.http.client.CookieStore;
030 import org.apache.http.client.methods.HttpPost;
031 import org.apache.http.client.params.ClientPNames;
032 import org.apache.http.client.params.CookiePolicy;
033 import org.apache.http.concurrent.FutureCallback;
034 import org.apache.http.entity.ByteArrayEntity;
035 import org.apache.http.impl.client.BasicCookieStore;
036 import org.apache.http.impl.nio.client.DefaultHttpAsyncClient;
037 import org.apache.http.nio.reactor.IOReactorStatus;
038 import org.apache.http.params.BasicHttpParams;
039 import org.granite.client.messaging.channel.Channel;
040 import org.granite.client.messaging.transport.AbstractTransport;
041 import org.granite.client.messaging.transport.HTTPTransport;
042 import org.granite.client.messaging.transport.TransportException;
043 import org.granite.client.messaging.transport.TransportFuture;
044 import org.granite.client.messaging.transport.TransportIOException;
045 import org.granite.client.messaging.transport.TransportMessage;
046 import org.granite.logging.Logger;
047 import org.granite.util.PublicByteArrayOutputStream;
048
049 /**
050 * @author Franck WOLFF
051 */
052 public class ApacheAsyncTransport extends AbstractTransport implements HTTPTransport {
053
054 private static final Logger log = Logger.getLogger(ApacheAsyncTransport.class);
055
056 protected final BasicHttpParams params;
057
058 protected DefaultHttpAsyncClient httpClient = null;
059 protected CookieStore cookieStore = new BasicCookieStore();
060
061 public ApacheAsyncTransport() {
062 this(null);
063 }
064
065 public ApacheAsyncTransport(BasicHttpParams params) {
066 this.params = params;
067 }
068
069 public void configure(DefaultHttpAsyncClient client) {
070 // Can be overwritten...
071 }
072
073 @Override
074 public synchronized boolean start() {
075 if (httpClient != null && httpClient.getStatus() == IOReactorStatus.ACTIVE)
076 return true;
077
078 stop();
079
080 log.info("Starting Apache HttpAsyncClient transport...");
081
082 try {
083 httpClient = new DefaultHttpAsyncClient();
084
085 configure(httpClient);
086 if (params != null)
087 params.copyParams(httpClient.getParams());
088
089 httpClient.setCookieStore(cookieStore);
090 httpClient.start();
091
092 final long timeout = System.currentTimeMillis() + 10000L; // 10sec.
093 while (httpClient.getStatus() != IOReactorStatus.ACTIVE) {
094 if (System.currentTimeMillis() > timeout)
095 throw new TimeoutException("HttpAsyncClient start process too long");
096 Thread.sleep(100);
097 }
098
099 log.info("Apache HttpAsyncClient transport started.");
100 return true;
101 }
102 catch (Exception e) {
103 httpClient = null;
104 getStatusHandler().handleException(new TransportException("Could not start Apache HttpAsyncClient", e));
105
106 log.error(e, "Apache HttpAsyncClient failed to start.");
107 return false;
108 }
109 }
110
111 @Override
112 public TransportFuture send(final Channel channel, final TransportMessage message) throws TransportException {
113 synchronized (this) {
114 if (httpClient == null && httpClient.getStatus() != IOReactorStatus.ACTIVE) {
115 TransportIOException e = new TransportIOException(message, "Apache HttpAsyncClient not started");
116 getStatusHandler().handleException(e);
117 throw e;
118 }
119 }
120
121 if (!message.isConnect())
122 getStatusHandler().handleIO(true);
123
124 try {
125 HttpPost request = new HttpPost(channel.getUri());
126 request.getParams().setParameter(ClientPNames.COOKIE_POLICY, CookiePolicy.BROWSER_COMPATIBILITY);
127 request.setHeader("Content-Type", message.getContentType());
128 request.setHeader("GDSClientType", "java"); // Notify the server that we expect Java serialization mode
129
130 PublicByteArrayOutputStream os = new PublicByteArrayOutputStream(512);
131 try {
132 message.encode(os);
133 }
134 catch (IOException e) {
135 throw new TransportException("Message serialization failed: " + message.getId(), e);
136 }
137 request.setEntity(new ByteArrayEntity(os.getBytes(), 0, os.size()));
138
139 // request.setEntity(new DeferredInputStreamEntity(message));
140
141 final Future<HttpResponse> future = httpClient.execute(request, new FutureCallback<HttpResponse>() {
142
143 public void completed(HttpResponse response) {
144 if (!message.isConnect())
145 getStatusHandler().handleIO(false);
146
147 InputStream is = null;
148 try {
149 is = response.getEntity().getContent();
150 channel.onMessage(is);
151 }
152 catch (Exception e) {
153 getStatusHandler().handleException(new TransportIOException(message, "Could not deserialize message", e));
154 }
155 finally {
156 if (is != null) try {
157 is.close();
158 }
159 catch (Exception e) {
160 }
161 }
162 }
163
164 public void failed(Exception e) {
165 if (!message.isConnect())
166 getStatusHandler().handleIO(false);
167
168 channel.onError(message, e);
169 getStatusHandler().handleException(new TransportIOException(message, "Request failed", e));
170 }
171
172 public void cancelled() {
173 if (!message.isConnect())
174 getStatusHandler().handleIO(false);
175
176 channel.onCancelled(message);
177 }
178 });
179
180 return new TransportFuture() {
181 @Override
182 public boolean cancel() {
183 boolean cancelled = false;
184 try {
185 cancelled = future.cancel(true);
186 }
187 catch (Exception e) {
188 log.error(e, "Cancel request failed");
189 }
190 return cancelled;
191 }
192 };
193 }
194 catch (Exception e) {
195 if (!message.isConnect())
196 getStatusHandler().handleIO(false);
197
198 TransportIOException f = new TransportIOException(message, "Request failed", e);
199 getStatusHandler().handleException(f);
200 throw f;
201 }
202 }
203
204 public synchronized void poll(final Channel channel, final TransportMessage message) throws TransportException {
205 throw new TransportException("Not implemented");
206 }
207
208 @Override
209 public synchronized void stop() {
210 if (httpClient == null)
211 return;
212
213 log.info("Stopping Apache HttpAsyncClient transport...");
214
215 super.stop();
216
217 try {
218 httpClient.shutdown();
219 }
220 catch (Exception e) {
221 getStatusHandler().handleException(new TransportException("Could not stop Apache HttpAsyncClient", e));
222
223 log.error(e, "Apache HttpAsyncClient failed to stop properly.");
224 }
225 finally {
226 httpClient = null;
227 }
228
229 log.info("Apache HttpAsyncClient transport stopped.");
230 }
231
232 // static class DeferredInputStreamEntity extends BasicHttpEntity {
233 //
234 // private final TransportMessage message;
235 // private InputStream content = null;
236 //
237 // public DeferredInputStreamEntity(TransportMessage message) {
238 // this.message = message;
239 // }
240 //
241 // @Override
242 // public synchronized InputStream getContent() throws IllegalStateException {
243 // if (content == null) {
244 // PublicByteArrayOutputStream os = new PublicByteArrayOutputStream(256);
245 // try {
246 // message.encode(os);
247 // }
248 // catch (IOException e) {
249 // throw new TransportException("Message serialization failed: " + message.getId(), e);
250 // }
251 // content = new ByteArrayInputStream(os.getBytes(), 0, os.size());
252 // }
253 // return content;
254 // }
255 // }
256 }