001 /*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.servicemix.http.processors;
018
019 import java.io.ByteArrayOutputStream;
020 import java.io.IOException;
021 import java.io.OutputStream;
022 import java.util.Enumeration;
023 import java.util.HashMap;
024 import java.util.Map;
025 import java.util.concurrent.ConcurrentHashMap;
026
027 import javax.jbi.messaging.DeliveryChannel;
028 import javax.jbi.messaging.ExchangeStatus;
029 import javax.jbi.messaging.Fault;
030 import javax.jbi.messaging.InOnly;
031 import javax.jbi.messaging.InOptionalOut;
032 import javax.jbi.messaging.InOut;
033 import javax.jbi.messaging.MessageExchange;
034 import javax.jbi.messaging.NormalizedMessage;
035 import javax.servlet.http.HttpServletRequest;
036
037 import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler;
038 import org.apache.commons.httpclient.Header;
039 import org.apache.commons.httpclient.HostConfiguration;
040 import org.apache.commons.httpclient.HttpClient;
041 import org.apache.commons.httpclient.HttpHost;
042 import org.apache.commons.httpclient.HttpMethod;
043 import org.apache.commons.httpclient.HttpStatus;
044 import org.apache.commons.httpclient.URI;
045 import org.apache.commons.httpclient.methods.ByteArrayRequestEntity;
046 import org.apache.commons.httpclient.methods.PostMethod;
047 import org.apache.commons.httpclient.methods.RequestEntity;
048 import org.apache.commons.httpclient.params.HttpMethodParams;
049 import org.apache.commons.httpclient.protocol.Protocol;
050 import org.apache.commons.httpclient.protocol.ProtocolSocketFactory;
051 import org.apache.commons.logging.Log;
052 import org.apache.commons.logging.LogFactory;
053 import org.apache.servicemix.common.JbiConstants;
054 import org.apache.servicemix.common.security.KeystoreManager;
055 import org.apache.servicemix.http.HttpComponent;
056 import org.apache.servicemix.http.HttpEndpoint;
057 import org.apache.servicemix.soap.Context;
058 import org.apache.servicemix.soap.SoapHelper;
059 import org.apache.servicemix.soap.SoapExchangeProcessor;
060 import org.apache.servicemix.soap.marshalers.SoapMessage;
061 import org.apache.servicemix.soap.marshalers.SoapReader;
062 import org.apache.servicemix.soap.marshalers.SoapWriter;
063
064 /**
065 *
066 * @author Guillaume Nodet
067 * @version $Revision: 370186 $
068 * @since 3.0
069 */
070 public class ProviderProcessor extends AbstractProcessor implements SoapExchangeProcessor {
071
072 private static Log log = LogFactory.getLog(ProviderProcessor.class);
073
074 protected SoapHelper soapHelper;
075 protected DeliveryChannel channel;
076 private Map<String, PostMethod> methods;
077 private Protocol protocol;
078
079 public ProviderProcessor(HttpEndpoint endpoint) {
080 super(endpoint);
081 this.soapHelper = new SoapHelper(endpoint);
082 this.methods = new ConcurrentHashMap<String, PostMethod>();
083 }
084
085 private String getRelUri(String locationUri) {
086 java.net.URI uri = java.net.URI.create(locationUri);
087 String relUri = uri.getPath();
088 if (!relUri.startsWith("/")) {
089 relUri = "/" + relUri;
090 }
091 if (uri.getQuery() != null) {
092 relUri += "?" + uri.getQuery();
093 }
094 if (uri.getFragment() != null) {
095 relUri += "#" + uri.getFragment();
096 }
097 return relUri;
098 }
099
100 public void process(MessageExchange exchange) throws Exception {
101 if (exchange.getStatus() == ExchangeStatus.DONE || exchange.getStatus() == ExchangeStatus.ERROR) {
102 PostMethod method = methods.remove(exchange.getExchangeId());
103 if (method != null) {
104 method.releaseConnection();
105 }
106 return;
107 }
108 boolean txSync = exchange.isTransacted() && Boolean.TRUE.equals(exchange.getProperty(JbiConstants.SEND_SYNC));
109 txSync |= endpoint.isSynchronous();
110 NormalizedMessage nm = exchange.getMessage("in");
111 if (nm == null) {
112 throw new IllegalStateException("Exchange has no input message");
113 }
114
115 String locationURI = endpoint.getLocationURI();
116
117 // Incorporated because of JIRA SM-695
118 Object newDestinationURI = nm.getProperty(JbiConstants.HTTP_DESTINATION_URI);
119 if (newDestinationURI != null) {
120 locationURI = (String) newDestinationURI;
121 log.debug("Location URI overridden: " + locationURI);
122 }
123
124 PostMethod method = new PostMethod(getRelUri(locationURI));
125 SoapMessage soapMessage = new SoapMessage();
126 soapHelper.getJBIMarshaler().fromNMS(soapMessage, nm);
127 Context context = soapHelper.createContext(soapMessage);
128 soapHelper.onSend(context);
129 SoapWriter writer = soapHelper.getSoapMarshaler().createWriter(soapMessage);
130 copyHeaderInformation(nm, method);
131 RequestEntity entity = writeMessage(writer);
132 // remove content-type header that may have been part of the in message
133 if (!endpoint.isWantContentTypeHeaderFromExchangeIntoHttpRequest()) {
134 method.removeRequestHeader(HEADER_CONTENT_TYPE);
135 method.addRequestHeader(HEADER_CONTENT_TYPE, entity.getContentType());
136 }
137 if (entity.getContentLength() < 0) {
138 method.removeRequestHeader(HEADER_CONTENT_LENGTH);
139 } else {
140 method.setRequestHeader(HEADER_CONTENT_LENGTH, Long.toString(entity.getContentLength()));
141 }
142 if (endpoint.isSoap() && method.getRequestHeader(HEADER_SOAP_ACTION) == null) {
143 if (endpoint.getSoapAction() != null) {
144 method.setRequestHeader(HEADER_SOAP_ACTION, endpoint.getSoapAction());
145 } else {
146 method.setRequestHeader(HEADER_SOAP_ACTION, "\"\"");
147 }
148 }
149 method.setRequestEntity(entity);
150 boolean close = true;
151 try {
152 // Set the retry handler
153 int retries = getConfiguration().isStreamingEnabled() ? 0 : getConfiguration().getRetryCount();
154 method.getParams().setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(retries, true));
155 // Set authentication
156 if (endpoint.getBasicAuthentication() != null) {
157 endpoint.getBasicAuthentication().applyCredentials(getClient(), exchange, nm);
158 }
159 // Execute the HTTP method
160 int response = getClient().executeMethod(getHostConfiguration(locationURI, exchange, nm), method);
161 if (response != HttpStatus.SC_OK && response != HttpStatus.SC_ACCEPTED) {
162 if (!(exchange instanceof InOnly)) {
163 SoapReader reader = soapHelper.getSoapMarshaler().createReader();
164 Header contentType = method.getResponseHeader(HEADER_CONTENT_TYPE);
165 soapMessage = reader.read(method.getResponseBodyAsStream(),
166 contentType != null ? contentType.getValue() : null);
167 context.setFaultMessage(soapMessage);
168 soapHelper.onAnswer(context);
169 Fault fault = exchange.createFault();
170 fault.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
171 soapHelper.getJBIMarshaler().toNMS(fault, soapMessage);
172 exchange.setFault(fault);
173 if (txSync) {
174 channel.sendSync(exchange);
175 } else {
176 methods.put(exchange.getExchangeId(), method);
177 channel.send(exchange);
178 close = false;
179 }
180 return;
181 } else {
182 throw new Exception("Invalid status response: " + response);
183 }
184 }
185 if (exchange instanceof InOut) {
186 close = processInOut(exchange, method, context, txSync, close);
187 } else if (exchange instanceof InOptionalOut) {
188 close = processInOptionalOut(method, exchange, context, txSync, close);
189 } else {
190 exchange.setStatus(ExchangeStatus.DONE);
191 channel.send(exchange);
192 }
193 } finally {
194 if (close) {
195 method.releaseConnection();
196 }
197 }
198 }
199
200 @SuppressWarnings("unchecked")
201 private void copyHeaderInformation(NormalizedMessage nm, PostMethod method) {
202 Map<String, String> headers = (Map<String, String>) nm.getProperty(JbiConstants.PROTOCOL_HEADERS);
203 if (headers != null) {
204 for (String name : headers.keySet()) {
205 String value = headers.get(name);
206 method.addRequestHeader(name, value);
207 }
208 }
209 }
210
211 private boolean processInOptionalOut(PostMethod method, MessageExchange exchange, Context context, boolean txSync,
212 boolean close) throws Exception {
213 if (method.getResponseContentLength() == 0) {
214 exchange.setStatus(ExchangeStatus.DONE);
215 channel.send(exchange);
216 } else {
217 NormalizedMessage msg = exchange.createMessage();
218 SoapReader reader = soapHelper.getSoapMarshaler().createReader();
219 SoapMessage soapMessage = reader.read(method.getResponseBodyAsStream(),
220 method.getResponseHeader(HEADER_CONTENT_TYPE).getValue());
221 context.setOutMessage(soapMessage);
222 soapHelper.onAnswer(context);
223 if (getConfiguration().isWantHeadersFromHttpIntoExchange()) {
224 msg.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
225 }
226 soapHelper.getJBIMarshaler().toNMS(msg, soapMessage);
227 ((InOptionalOut) exchange).setOutMessage(msg);
228 if (txSync) {
229 channel.sendSync(exchange);
230 } else {
231 methods.put(exchange.getExchangeId(), method);
232 channel.send(exchange);
233 close = false;
234 }
235 }
236 return close;
237 }
238
239 private boolean processInOut(MessageExchange exchange, PostMethod method, Context context, boolean txSync,
240 boolean close) throws Exception {
241 NormalizedMessage msg = exchange.createMessage();
242 SoapReader reader = soapHelper.getSoapMarshaler().createReader();
243 Header contentType = method.getResponseHeader(HEADER_CONTENT_TYPE);
244 SoapMessage soapMessage = reader.read(method.getResponseBodyAsStream(), contentType != null ? contentType.getValue() : null);
245 context.setOutMessage(soapMessage);
246 soapHelper.onAnswer(context);
247 if (getConfiguration().isWantHeadersFromHttpIntoExchange()) {
248 msg.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
249 }
250 soapHelper.getJBIMarshaler().toNMS(msg, soapMessage);
251 ((InOut) exchange).setOutMessage(msg);
252 if (txSync) {
253 channel.sendSync(exchange);
254 } else {
255 methods.put(exchange.getExchangeId(), method);
256 channel.send(exchange);
257 close = false;
258 }
259 return close;
260 }
261
262 private HostConfiguration getHostConfiguration(String locationURI, MessageExchange exchange, NormalizedMessage message)
263 throws Exception {
264 HostConfiguration host;
265 URI uri = new URI(locationURI, false);
266 if (uri.getScheme().equals("https")) {
267 synchronized (this) {
268 if (protocol == null) {
269 ProtocolSocketFactory sf = new CommonsHttpSSLSocketFactory(
270 endpoint.getSsl(),
271 KeystoreManager.Proxy.create(endpoint.getKeystoreManager()));
272 protocol = new Protocol("https", sf, 443);
273 }
274 }
275 HttpHost httphost = new HttpHost(uri.getHost(), uri.getPort(), protocol);
276 host = new HostConfiguration();
277 host.setHost(httphost);
278 } else {
279 host = new HostConfiguration();
280 host.setHost(uri.getHost(), uri.getPort());
281 }
282 if (endpoint.getProxy() != null) {
283 if ((endpoint.getProxy().getProxyHost() != null) && (endpoint.getProxy().getProxyPort() != 0)) {
284 host.setProxy(endpoint.getProxy().getProxyHost(), endpoint.getProxy().getProxyPort());
285 }
286 if (endpoint.getProxy().getProxyCredentials() != null) {
287 endpoint.getProxy().getProxyCredentials().applyProxyCredentials(getClient(), exchange, message);
288 }
289 } else if ((getConfiguration().getProxyHost() != null) && (getConfiguration().getProxyPort() != 0)) {
290 host.setProxy(getConfiguration().getProxyHost(), getConfiguration().getProxyPort());
291 }
292 return host;
293 }
294
295 public void init() throws Exception {
296 channel = endpoint.getServiceUnit().getComponent().getComponentContext().getDeliveryChannel();
297 }
298
299 public void start() throws Exception {
300 }
301
302 public void stop() throws Exception {
303 }
304
305 public void shutdown() throws Exception {
306 }
307
308 protected Map<String, String> getHeaders(HttpServletRequest request) {
309 Map<String, String> headers = new HashMap<String, String>();
310 Enumeration<?> enumeration = request.getHeaderNames();
311 while (enumeration.hasMoreElements()) {
312 String name = (String) enumeration.nextElement();
313 String value = request.getHeader(name);
314 headers.put(name, value);
315 }
316 return headers;
317 }
318
319 protected Map<String, String> getHeaders(HttpMethod method) {
320 Map<String, String> headers = new HashMap<String, String>();
321 Header[] h = method.getResponseHeaders();
322 for (int i = 0; i < h.length; i++) {
323 headers.put(h[i].getName(), h[i].getValue());
324 }
325 return headers;
326 }
327
328 protected RequestEntity writeMessage(SoapWriter writer) throws Exception {
329 if (getConfiguration().isStreamingEnabled()) {
330 return new StreamingRequestEntity(writer);
331 } else {
332 ByteArrayOutputStream baos = new ByteArrayOutputStream();
333 writer.write(baos);
334 return new ByteArrayRequestEntity(baos.toByteArray(), writer.getContentType());
335 }
336 }
337
338 protected HttpClient getClient() {
339 HttpComponent comp = (HttpComponent) endpoint.getServiceUnit().getComponent();
340 HttpClient client = comp.getClient();
341 client.getParams().setSoTimeout(endpoint.getTimeout());
342 return client;
343 }
344
345 public static class StreamingRequestEntity implements RequestEntity {
346
347 private SoapWriter writer;
348
349 public StreamingRequestEntity(SoapWriter writer) {
350 this.writer = writer;
351 }
352
353 public boolean isRepeatable() {
354 return false;
355 }
356
357 public void writeRequest(OutputStream out) throws IOException {
358 try {
359 writer.write(out);
360 out.flush();
361 } catch (Exception e) {
362 throw (IOException) new IOException("Could not write request").initCause(e);
363 }
364 }
365
366 public long getContentLength() {
367 // not known so we send negative value
368 return -1;
369 }
370
371 public String getContentType() {
372 return writer.getContentType();
373 }
374
375 }
376 }