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.endpoints;
018
019 import java.io.IOException;
020 import java.net.URI;
021 import java.util.HashMap;
022 import java.util.Map;
023 import java.util.concurrent.ConcurrentHashMap;
024
025 import javax.jbi.management.DeploymentException;
026 import javax.jbi.messaging.ExchangeStatus;
027 import javax.jbi.messaging.Fault;
028 import javax.jbi.messaging.MessageExchange;
029 import javax.jbi.messaging.NormalizedMessage;
030 import javax.jbi.servicedesc.ServiceEndpoint;
031 import javax.security.auth.Subject;
032 import javax.servlet.ServletException;
033 import javax.servlet.http.HttpServletRequest;
034 import javax.servlet.http.HttpServletResponse;
035 import javax.xml.namespace.QName;
036 import javax.xml.transform.TransformerException;
037 import javax.xml.transform.dom.DOMSource;
038 import javax.xml.transform.stream.StreamResult;
039
040 import org.w3c.dom.Node;
041
042 import org.apache.servicemix.common.DefaultComponent;
043 import org.apache.servicemix.common.ServiceUnit;
044 import org.apache.servicemix.common.JbiConstants;
045 import org.apache.servicemix.common.endpoints.ConsumerEndpoint;
046 import org.apache.servicemix.http.ContextManager;
047 import org.apache.servicemix.http.HttpComponent;
048 import org.apache.servicemix.http.HttpEndpointType;
049 import org.apache.servicemix.http.HttpProcessor;
050 import org.apache.servicemix.http.SslParameters;
051 import org.apache.servicemix.http.jetty.JaasJettyPrincipal;
052 import org.apache.servicemix.jbi.jaxp.SourceTransformer;
053 import org.mortbay.jetty.RetryRequest;
054 import org.mortbay.util.ajax.Continuation;
055 import org.mortbay.util.ajax.ContinuationSupport;
056
057 /**
058 * Plain HTTP consumer endpoint. This endpoint can be used to handle plain HTTP request (without SOAP) or to be able to
059 * process the request in a non standard way. For HTTP requests, a WSDL2 HTTP binding can be used.
060 *
061 * @author gnodet
062 * @since 3.2
063 * @org.apache.xbean.XBean element="consumer"
064 */
065 public class HttpConsumerEndpoint extends ConsumerEndpoint implements HttpProcessor, HttpEndpointType {
066
067 public static final String MAIN_WSDL = "main.wsdl";
068
069 private String authMethod;
070 private SslParameters ssl;
071 private String locationURI;
072 private HttpConsumerMarshaler marshaler;
073 private long timeout; // 0 => default to the timeout configured on component
074 private URI defaultMep = JbiConstants.IN_OUT;
075
076 private Map<String, Object> resources = new HashMap<String, Object>();
077 private Map<String, Continuation> locks = new ConcurrentHashMap<String, Continuation>();
078 private Map<String, MessageExchange> exchanges = new ConcurrentHashMap<String, MessageExchange>();
079 private Object httpContext;
080
081 private boolean started = false;
082
083 public HttpConsumerEndpoint() {
084 super();
085 }
086
087 public HttpConsumerEndpoint(DefaultComponent component, ServiceEndpoint endpoint) {
088 super(component, endpoint);
089 }
090
091 public HttpConsumerEndpoint(ServiceUnit serviceUnit, QName service, String endpoint) {
092 super(serviceUnit, service, endpoint);
093 }
094
095 /**
096 * Returns the URI at which the endpoint listens for new requests.
097 *
098 * @return a string representing the endpoint's URI
099 */
100 public String getLocationURI() {
101 return locationURI;
102 }
103
104 /**
105 * Sets the URI at which an endpoint listens for requests.
106 *
107 * @param locationURI a string representing the URI
108 * @org.apache.xbean.Property description="the URI at which the endpoint listens for requests"
109 */
110 public void setLocationURI(String locationURI) {
111 this.locationURI = locationURI;
112 }
113
114 /**
115 * Returns the timeout value for an HTTP endpoint.
116 *
117 * @return the timeout specified in milliseconds
118 */
119 public long getTimeout() {
120 return timeout;
121 }
122
123 /**
124 * Specifies the timeout value for an HTTP consumer endpoint. The timeout is specified in milliseconds. The default value is 0
125 * which means that the endpoint will never timeout.
126 *
127 * @org.apache.xbean.Property description="the timeout is specified in milliseconds. The default value is 0 which
128 * means that the endpoint will never timeout."
129 * @param timeout the length time, in milliseconds, to wait before timing out
130 */
131 public void setTimeout(long timeout) {
132 this.timeout = timeout;
133 }
134
135 /**
136 * @return the marshaler
137 */
138 public HttpConsumerMarshaler getMarshaler() {
139 return marshaler;
140 }
141
142 /**
143 * Sets the class used to marshal messages.
144 *
145 * @param marshaler the marshaler to set
146 * @org.apache.xbean.Property description="the bean used to marshal HTTP messages. The default is a
147 * <code>DefaultHttpConsumerMarshaler</code>."
148 */
149 public void setMarshaler(HttpConsumerMarshaler marshaler) {
150 this.marshaler = marshaler;
151 }
152
153 /**
154 * Returns a string describing the authentication scheme being used by an endpoint.
155 *
156 * @return a string representing the authentication method used by an endpoint
157 */
158 public String getAuthMethod() {
159 return authMethod;
160 }
161
162 /**
163 * Specifies the authentication method used by a secure endpoint. The authentication method is a string naming the scheme used
164 * for authenticating users.
165 *
166 * @param authMethod a string naming the authentication scheme a secure endpoint should use
167 * @org.apache.xbean.Property description="a string naming the scheme used for authenticating users"
168 */
169 public void setAuthMethod(String authMethod) {
170 this.authMethod = authMethod;
171 }
172
173 /**
174 * @return the sslParameters
175 */
176 public SslParameters getSsl() {
177 return ssl;
178 }
179
180 /**
181 * Sets the properties used to configure SSL for the endpoint.
182 *
183 * @param ssl an <code>SslParameters</code> object containing the SSL properties
184 * @org.apache.xbean.Property description="a bean containing the SSL configuration properties"
185 */
186 public void setSsl(SslParameters ssl) {
187 this.ssl = ssl;
188 }
189
190 /**
191 * Returns a URI representing the default message exachange pattern(MEP) used by an endpoint.
192 *
193 * @return a URI representing an endpoint's default MEP
194 */
195 public URI getDefaultMep() {
196 return defaultMep;
197 }
198
199 /**
200 * Sets the default message exchange pattern(MEP) for an endpoint. The default MEP is specified as a URI and the default is
201 * <code>JbiConstants.IN_OUT</code>.
202 *
203 * @param defaultMep a URI representing the default MEP of the endpoint
204 * @org.apache.xbean.Property description="a URI representing the endpoint's default MEP. The default is
205 * <code>JbiConstants.IN_OUT</code>."
206 */
207 public void setDefaultMep(URI defaultMep) {
208 this.defaultMep = defaultMep;
209 }
210
211 public void activate() throws Exception {
212 super.activate();
213 loadStaticResources();
214 httpContext = getServerManager().createContext(locationURI, this);
215 }
216
217 public void deactivate() throws Exception {
218 getServerManager().remove(httpContext);
219 httpContext = null;
220 super.deactivate();
221 }
222
223 public void start() throws Exception {
224 super.start();
225 started = true;
226 }
227
228 public void stop() throws Exception {
229 started = false;
230 super.stop();
231 }
232
233 public void process(MessageExchange exchange) throws Exception {
234 // Receive the exchange response
235 // First, check if the continuation has not been removed from the map,
236 // which would mean it has timed out. If this is the case, throw an exception
237 // that will set the exchange status to ERROR.
238 Continuation cont = locks.get(exchange.getExchangeId());
239 if (cont == null) {
240 throw new Exception("HTTP request has timed out for exchange: " + exchange.getExchangeId());
241 }
242 // synchronized block
243 synchronized (cont) {
244 if (locks.remove(exchange.getExchangeId()) == null) {
245 throw new Exception("HTTP request has timed out for exchange: " + exchange.getExchangeId());
246 }
247 if (logger.isDebugEnabled()) {
248 logger.debug("Resuming continuation for exchange: " + exchange.getExchangeId());
249 }
250 // Put the new exchange
251 exchanges.put(exchange.getExchangeId(), exchange);
252 // Resume continuation
253 cont.resume();
254 if (!cont.isResumed()) {
255 if (logger.isDebugEnabled()) {
256 logger.debug("Could not resume continuation for exchange: " + exchange.getExchangeId());
257 }
258 exchanges.remove(exchange.getExchangeId());
259 throw new Exception("HTTP request has timed out for exchange: " + exchange.getExchangeId());
260 }
261 }
262 }
263
264 public void process(HttpServletRequest request, HttpServletResponse response) throws Exception {
265 if (logger.isDebugEnabled()) {
266 logger.debug("Receiving HTTP request: " + request);
267 }
268 MessageExchange exchange = null;
269 try {
270 // Handle WSDLs, XSDs
271 if (handleStaticResource(request, response)) {
272 return;
273 }
274 // Check endpoint is started
275 if (!started) {
276 response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Endpoint is stopped");
277 return;
278 }
279 // Not giving a specific mutex will synchronize on the continuation
280 // itself
281 Continuation cont = ContinuationSupport.getContinuation(request, null);
282 // If the continuation is not a retry
283 if (!cont.isPending()) {
284 // Create the exchange
285 exchange = createExchange(request);
286 // Put the exchange in a map so that we can later retrieve it
287 // We don't put the exchange on the request directly in case the JMS flow is involved
288 // because the exchange coming back may not be the same object as the one send.
289 exchanges.put(exchange.getExchangeId(), exchange);
290 // Put the exchange id on the request to be able to retrieve the exchange later
291 request.setAttribute(MessageExchange.class.getName(), exchange.getExchangeId());
292 // Put the continuation in a map under the exchange id key
293 locks.put(exchange.getExchangeId(), cont);
294 synchronized (cont) {
295 // Send the exchange
296 send(exchange);
297 if (logger.isDebugEnabled()) {
298 logger.debug("Suspending continuation for exchange: " + exchange.getExchangeId());
299 }
300 // Suspend the continuation for the configured timeout
301 // If a SelectConnector is used, the call to suspend will throw a RetryRequest exception
302 // else, the call will block until the continuation is resumed
303 long to = this.timeout;
304 if (to == 0) {
305 to = ((HttpComponent) getServiceUnit().getComponent()).getConfiguration().getConsumerProcessorSuspendTime();
306 }
307 boolean result = cont.suspend(to);
308 // The call has not thrown a RetryRequest, which means we don't use a SelectConnector
309 // and we must handle the exchange in this very method call.
310 // If result is false, the continuation has timed out.
311 // So get the exchange (in case the object has changed) and remove it from the map
312 exchange = exchanges.remove(exchange.getExchangeId());
313 // remove the exchange id from the request as we don't need it anymore
314 request.removeAttribute(MessageExchange.class.getName());
315 // If a timeout occurred, throw an exception that will be sent back to the HTTP client
316 // Whenever the exchange comes back, the process(MessageExchange) method will thrown an
317 // exception and the exchange will be set in an ERROR status
318 if (!result) {
319 // Remove the continuation from the map.
320 // This indicates the continuation has been fully processed
321 locks.remove(exchange.getExchangeId());
322 throw new Exception("Exchange timed out");
323 }
324 }
325 // The continuation is a retry.
326 // This happens when the SelectConnector is used and in two cases:
327 // * the continuation has been resumed because the exchange has been received
328 // * the continuation has timed out
329 } else {
330 synchronized (cont) {
331 // Get the exchange id from the request
332 String id = (String) request.getAttribute(MessageExchange.class.getName());
333 // Remove the continuation from the map, indicating it has been processed or timed out
334 locks.remove(id);
335 exchange = exchanges.remove(id);
336 request.removeAttribute(MessageExchange.class.getName());
337 // Check if this is a timeout
338 if (exchange == null) {
339 throw new IllegalStateException("Exchange not found");
340 }
341 if (!cont.isResumed()) {
342 // When the exchange comes back later, the continuation will not be found and
343 // the exchange will be set in an ERROR state by the process(MessageExchange) method
344 throw new Exception("Exchange timed out: " + exchange.getExchangeId());
345 }
346 }
347 }
348 // At this point, we have received the exchange response,
349 // so process it and send back the HTTP response
350 if (exchange.getStatus() == ExchangeStatus.ERROR) {
351 Exception e = exchange.getError();
352 if (e == null) {
353 e = new Exception("Unkown error (exchange aborted ?)");
354 }
355 throw e;
356 } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
357 try {
358 Fault fault = exchange.getFault();
359 if (fault != null) {
360 sendFault(exchange, fault, request, response);
361 } else {
362 NormalizedMessage outMsg = exchange.getMessage("out");
363 if (outMsg != null) {
364 sendOut(exchange, outMsg, request, response);
365 }
366 }
367 done(exchange);
368 } catch (Exception e) {
369 fail(exchange, e);
370 throw e;
371 }
372 } else if (exchange.getStatus() == ExchangeStatus.DONE) {
373 // This happens when there is no response to send back
374 sendAccepted(exchange, request, response);
375 }
376 } catch (RetryRequest e) {
377 throw e;
378 } catch (Exception e) {
379 sendError(exchange, e, request, response);
380 }
381 }
382
383 protected void loadStaticResources() {
384 // TODO: load wsdl
385 }
386
387 /**
388 * Handle static resources
389 *
390 * @param request the http request
391 * @param response the http response
392 * @return <code>true</code> if the request has been handled
393 * @throws IOException
394 * @throws ServletException
395 */
396 protected boolean handleStaticResource(HttpServletRequest request, HttpServletResponse response)
397 throws IOException, ServletException {
398 if (!"GET".equals(request.getMethod())) {
399 return false;
400 }
401 String query = request.getQueryString();
402 if (query != null && query.trim().equalsIgnoreCase("wsdl") && getResource(MAIN_WSDL) != null) {
403 String uri = request.getRequestURI();
404 if (!uri.endsWith("/")) {
405 uri += "/";
406 }
407 uri += MAIN_WSDL;
408 response.sendRedirect(uri);
409 return true;
410 }
411 String path = request.getPathInfo();
412 if (path.lastIndexOf('/') >= 0) {
413 path = path.substring(path.lastIndexOf('/') + 1);
414 }
415 Object res = getResource(path);
416 if (res == null) {
417 return false;
418 }
419 if (res instanceof Node) {
420 response.setStatus(200);
421 response.setContentType("text/xml");
422 try {
423 new SourceTransformer().toResult(new DOMSource((Node)res),
424 new StreamResult(response.getOutputStream()));
425 } catch (TransformerException e) {
426 throw new ServletException("Error while sending xml resource", e);
427 }
428 } else if (res != null) {
429 // TODO: handle other static resources ...
430 throw new ServletException("Unable to serialize resource");
431 } else {
432 return false;
433 }
434 return true;
435 }
436
437 protected Object getResource(String path) {
438 return resources.get(path);
439 }
440
441 protected void addResource(String path, Object resource) {
442 resources.put(path, resource);
443 }
444
445 protected ContextManager getServerManager() {
446 HttpComponent comp = (HttpComponent) getServiceUnit().getComponent();
447 return comp.getServer();
448 }
449
450 public MessageExchange createExchange(HttpServletRequest request) throws Exception {
451 MessageExchange me = marshaler.createExchange(request, getContext());
452 configureExchangeTarget(me);
453 // If the user has been authenticated, put these informations on
454 // the in NormalizedMessage.
455 if (request.getUserPrincipal() instanceof JaasJettyPrincipal) {
456 Subject subject = ((JaasJettyPrincipal) request.getUserPrincipal()).getSubject();
457 me.getMessage("in").setSecuritySubject(subject);
458 }
459 return me;
460 }
461
462 public void sendAccepted(MessageExchange exchange, HttpServletRequest request,
463 HttpServletResponse response) throws Exception {
464 marshaler.sendAccepted(exchange, request, response);
465 }
466
467 public void sendError(MessageExchange exchange, Exception error, HttpServletRequest request,
468 HttpServletResponse response) throws Exception {
469 marshaler.sendError(exchange, error, request, response);
470 }
471
472 public void sendFault(MessageExchange exchange, Fault fault, HttpServletRequest request,
473 HttpServletResponse response) throws Exception {
474 marshaler.sendFault(exchange, fault, request, response);
475 }
476
477 public void sendOut(MessageExchange exchange, NormalizedMessage outMsg, HttpServletRequest request,
478 HttpServletResponse response) throws Exception {
479 marshaler.sendOut(exchange, outMsg, request, response);
480 }
481
482 public void validate() throws DeploymentException {
483 super.validate();
484 if (marshaler == null) {
485 marshaler = new DefaultHttpConsumerMarshaler();
486 }
487 if (marshaler instanceof DefaultHttpConsumerMarshaler) {
488 ((DefaultHttpConsumerMarshaler) marshaler).setDefaultMep(getDefaultMep());
489 }
490 }
491 }