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.net.URI;
020 import java.util.Enumeration;
021 import java.util.HashMap;
022 import java.util.Map;
023 import java.util.concurrent.ConcurrentHashMap;
024
025 import javax.jbi.component.ComponentContext;
026 import javax.jbi.messaging.DeliveryChannel;
027 import javax.jbi.messaging.ExchangeStatus;
028 import javax.jbi.messaging.MessageExchange;
029 import javax.jbi.messaging.NormalizedMessage;
030 import javax.security.auth.Subject;
031 import javax.servlet.http.HttpServletRequest;
032 import javax.servlet.http.HttpServletResponse;
033 import javax.xml.namespace.QName;
034 import javax.xml.transform.dom.DOMSource;
035 import javax.xml.transform.stream.StreamResult;
036
037 import org.w3c.dom.Node;
038
039 import org.apache.commons.logging.Log;
040 import org.apache.commons.logging.LogFactory;
041 import org.apache.servicemix.common.JbiConstants;
042 import org.apache.servicemix.common.EndpointComponentContext;
043 import org.apache.servicemix.http.ContextManager;
044 import org.apache.servicemix.http.HttpComponent;
045 import org.apache.servicemix.http.HttpEndpoint;
046 import org.apache.servicemix.http.HttpProcessor;
047 import org.apache.servicemix.http.SslParameters;
048 import org.apache.servicemix.http.jetty.JaasJettyPrincipal;
049 import org.apache.servicemix.jbi.jaxp.SourceTransformer;
050 import org.apache.servicemix.soap.Context;
051 import org.apache.servicemix.soap.SoapFault;
052 import org.apache.servicemix.soap.SoapHelper;
053 import org.apache.servicemix.soap.SoapExchangeProcessor;
054 import org.apache.servicemix.soap.marshalers.JBIMarshaler;
055 import org.apache.servicemix.soap.marshalers.SoapMessage;
056 import org.apache.servicemix.soap.marshalers.SoapWriter;
057 import org.mortbay.jetty.RetryRequest;
058 import org.mortbay.util.ajax.Continuation;
059 import org.mortbay.util.ajax.ContinuationSupport;
060
061 public class ConsumerProcessor extends AbstractProcessor implements SoapExchangeProcessor, HttpProcessor {
062
063 private static Log log = LogFactory.getLog(ConsumerProcessor.class);
064
065 protected Object httpContext;
066 protected ComponentContext context;
067 protected DeliveryChannel channel;
068 protected SoapHelper soapHelper;
069 protected Map<String, Continuation> locks;
070 protected Map<String, MessageExchange> exchanges;
071 protected int suspentionTime = 60000;
072 protected boolean started = false;
073
074 public ConsumerProcessor(HttpEndpoint endpoint) {
075 super(endpoint);
076 this.soapHelper = new SoapHelper(endpoint);
077 this.locks = new ConcurrentHashMap<String, Continuation>();
078 this.exchanges = new ConcurrentHashMap<String, MessageExchange>();
079 this.suspentionTime = endpoint.getTimeout();
080 if (suspentionTime <= 0) {
081 this.suspentionTime = getConfiguration().getConsumerProcessorSuspendTime();
082 }
083 }
084
085 public SslParameters getSsl() {
086 return this.endpoint.getSsl();
087 }
088
089 public String getAuthMethod() {
090 return this.endpoint.getAuthMethod();
091 }
092
093 public void process(MessageExchange exchange) throws Exception {
094 Continuation cont = locks.get(exchange.getExchangeId());
095 if (cont == null) {
096 throw new Exception("HTTP request has timed out");
097 }
098 synchronized (cont) {
099 if (locks.remove(exchange.getExchangeId()) == null) {
100 throw new Exception("HTTP request has timed out");
101 }
102 if (log.isDebugEnabled()) {
103 log.debug("Resuming continuation for exchange: " + exchange.getExchangeId());
104 }
105 exchanges.put(exchange.getExchangeId(), exchange);
106 cont.resume();
107 if (!cont.isResumed()) {
108 if (log.isDebugEnabled()) {
109 log.debug("Could not resume continuation for exchange: " + exchange.getExchangeId());
110 }
111 exchanges.remove(exchange.getExchangeId());
112 throw new Exception("HTTP request has timed out for exchange: " + exchange.getExchangeId());
113 }
114 }
115 }
116
117 public void init() throws Exception {
118 String url = endpoint.getLocationURI();
119 context = new EndpointComponentContext(endpoint);
120 channel = context.getDeliveryChannel();
121 httpContext = getServerManager().createContext(url, this);
122 }
123
124 public void shutdown() throws Exception {
125 getServerManager().remove(httpContext);
126 }
127
128 public void start() throws Exception {
129 started = true;
130 }
131
132 public void stop() throws Exception {
133 started = false;
134 }
135
136 public void process(HttpServletRequest request, HttpServletResponse response) throws Exception {
137 if (log.isDebugEnabled()) {
138 log.debug("Receiving HTTP request: " + request);
139 }
140 if ("GET".equals(request.getMethod())) {
141 processGetRequest(request, response);
142 return;
143 }
144 if (!started) {
145 response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE, "Endpoint is stopped");
146 return;
147 }
148 if (!"POST".equals(request.getMethod())) {
149 response.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED, request.getMethod() + " not supported");
150 return;
151 }
152 // Not giving a specific mutex will synchronize on the contination itself
153 Continuation cont = ContinuationSupport.getContinuation(request, null);
154 MessageExchange exchange;
155 // If the continuation is not a retry
156 if (!cont.isPending()) {
157 try {
158 Context ctx = createContext(request);
159 request.setAttribute(Context.class.getName(), ctx);
160 exchange = soapHelper.onReceive(ctx);
161 exchanges.put(exchange.getExchangeId(), exchange);
162 NormalizedMessage inMessage = exchange.getMessage("in");
163 if (getConfiguration().isWantHeadersFromHttpIntoExchange()) {
164 inMessage.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(request));
165 }
166 locks.put(exchange.getExchangeId(), cont);
167 request.setAttribute(MessageExchange.class.getName(), exchange.getExchangeId());
168 synchronized (cont) {
169 channel.send(exchange);
170 if (log.isDebugEnabled()) {
171 log.debug("Suspending continuation for exchange: " + exchange.getExchangeId());
172 }
173 boolean result = cont.suspend(suspentionTime);
174 exchange = exchanges.remove(exchange.getExchangeId());
175 request.removeAttribute(MessageExchange.class.getName());
176 if (!result) {
177 locks.remove(exchange.getExchangeId());
178 throw new Exception("Exchange timed out");
179 }
180 }
181 } catch (RetryRequest retry) {
182 throw retry;
183 } catch (SoapFault fault) {
184 sendFault(fault, request, response);
185 return;
186 } catch (Exception e) {
187 sendFault(new SoapFault(e), request, response);
188 return;
189 }
190 } else {
191 synchronized (cont) {
192 String id = (String) request.getAttribute(MessageExchange.class.getName());
193 locks.remove(id);
194 exchange = exchanges.remove(id);
195 request.removeAttribute(MessageExchange.class.getName());
196 // Check if this is a timeout
197 if (exchange == null) {
198 throw new IllegalStateException("Exchange not found");
199 }
200 if (!cont.isResumed()) {
201 Exception e = new Exception("Exchange timed out: " + exchange.getExchangeId());
202 sendFault(new SoapFault(e), request, response);
203 return;
204 }
205 }
206 }
207 if (exchange.getStatus() == ExchangeStatus.ERROR) {
208 if (exchange.getError() != null) {
209 throw new Exception(exchange.getError());
210 } else {
211 throw new Exception("Unknown Error");
212 }
213 } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
214 try {
215 if (exchange.getFault() != null) {
216 processFault(exchange, request, response);
217 } else {
218 processResponse(exchange, request, response);
219 }
220 } finally {
221 exchange.setStatus(ExchangeStatus.DONE);
222 channel.send(exchange);
223 }
224 } else if (exchange.getStatus() == ExchangeStatus.DONE) {
225 // This happens when there is no response to send back
226 response.setStatus(HttpServletResponse.SC_ACCEPTED);
227 }
228 }
229
230 private Context createContext(HttpServletRequest request) throws Exception {
231 SoapMessage message = soapHelper.getSoapMarshaler().createReader().read(
232 request.getInputStream(),
233 request.getHeader(HEADER_CONTENT_TYPE));
234 Context ctx = soapHelper.createContext(message);
235 if (request.getUserPrincipal() != null) {
236 if (request.getUserPrincipal() instanceof JaasJettyPrincipal) {
237 Subject subject = ((JaasJettyPrincipal) request.getUserPrincipal()).getSubject();
238 ctx.getInMessage().setSubject(subject);
239 } else {
240 ctx.getInMessage().addPrincipal(request.getUserPrincipal());
241 }
242 }
243 return ctx;
244 }
245
246 private void processResponse(MessageExchange exchange, HttpServletRequest request, HttpServletResponse response) throws Exception {
247 NormalizedMessage outMsg = exchange.getMessage("out");
248 if (outMsg != null) {
249 Context ctx = (Context) request.getAttribute(Context.class.getName());
250 SoapMessage out = soapHelper.onReply(ctx, outMsg);
251 SoapWriter writer = soapHelper.getSoapMarshaler().createWriter(out);
252 response.setContentType(writer.getContentType());
253 writer.write(response.getOutputStream());
254 }
255 }
256
257 private void processFault(MessageExchange exchange, HttpServletRequest request, HttpServletResponse response) throws Exception {
258 SoapFault fault = new SoapFault(
259 (QName) exchange.getFault().getProperty(JBIMarshaler.SOAP_FAULT_CODE),
260 (QName) exchange.getFault().getProperty(JBIMarshaler.SOAP_FAULT_SUBCODE),
261 (String) exchange.getFault().getProperty(JBIMarshaler.SOAP_FAULT_REASON),
262 (URI) exchange.getFault().getProperty(JBIMarshaler.SOAP_FAULT_NODE),
263 (URI) exchange.getFault().getProperty(JBIMarshaler.SOAP_FAULT_ROLE),
264 exchange.getFault().getContent());
265 sendFault(fault, request, response);
266 }
267
268 private void processGetRequest(HttpServletRequest request, HttpServletResponse response) throws Exception {
269 String query = request.getQueryString();
270 if (query != null && query.trim().equalsIgnoreCase("wsdl")) {
271 String uri = request.getRequestURI();
272 if (!uri.endsWith("/")) {
273 uri += "/";
274 }
275 uri += "main.wsdl";
276 response.sendRedirect(uri);
277 return;
278 }
279 String path = request.getPathInfo();
280 if (path.lastIndexOf('/') >= 0) {
281 path = path.substring(path.lastIndexOf('/') + 1);
282 }
283
284 // Set protocol, host, and port in the component
285 HttpComponent comp = (HttpComponent) endpoint.getServiceUnit().getComponent();
286 comp.setProtocol(request.getScheme());
287 comp.setHost(request.getServerName());
288 comp.setPort(request.getServerPort());
289 comp.setPath(request.getContextPath());
290
291 // Reload the wsdl
292 endpoint.reloadWsdl();
293
294 Node node = (Node) endpoint.getWsdls().get(path);
295 generateDocument(response, node);
296 }
297
298 protected void sendFault(SoapFault fault, HttpServletRequest request, HttpServletResponse response) throws Exception {
299 if (SoapFault.SENDER.equals(fault.getCode())) {
300 response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
301 } else {
302 response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
303 }
304 Context ctx = (Context) request.getAttribute(Context.class.getName());
305 SoapMessage soapFault = soapHelper.onFault(ctx, fault);
306 SoapWriter writer = soapHelper.getSoapMarshaler().createWriter(soapFault);
307 response.setContentType(writer.getContentType());
308 writer.write(response.getOutputStream());
309 }
310
311 protected Map<String, String> getHeaders(HttpServletRequest request) {
312 Map<String, String> headers = new HashMap<String, String>();
313 Enumeration<?> enumeration = request.getHeaderNames();
314 while (enumeration.hasMoreElements()) {
315 String name = (String) enumeration.nextElement();
316 String value = request.getHeader(name);
317 headers.put(name, value);
318 }
319 return headers;
320 }
321
322 protected ContextManager getServerManager() {
323 HttpComponent comp = (HttpComponent) endpoint.getServiceUnit().getComponent();
324 return comp.getServer();
325 }
326
327 protected void generateDocument(HttpServletResponse response, Node node) throws Exception {
328 if (node == null) {
329 response.sendError(HttpServletResponse.SC_NOT_FOUND, "Unable to find requested resource");
330 return;
331 }
332 response.setStatus(200);
333 response.setContentType("text/xml");
334 new SourceTransformer().toResult(new DOMSource(node), new StreamResult(response.getOutputStream()));
335 }
336
337 }