You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicemix.apache.org by gn...@apache.org on 2006/01/18 18:01:54 UTC
svn commit: r370186 - in
/incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm:
./ ConsumerProcessor.java ProviderProcessor.java
Author: gnodet
Date: Wed Jan 18 09:01:53 2006
New Revision: 370186
URL: http://svn.apache.org/viewcvs?rev=370186&view=rev
Log:
Add missing classes
Added:
incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/
incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ConsumerProcessor.java
incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ProviderProcessor.java
- copied, changed from r367101, incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/ProviderProcessor.java
Added: incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ConsumerProcessor.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ConsumerProcessor.java?rev=370186&view=auto
==============================================================================
--- incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ConsumerProcessor.java (added)
+++ incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ConsumerProcessor.java Wed Jan 18 09:01:53 2006
@@ -0,0 +1,149 @@
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.servicemix.http.processors;
+
+import java.net.URI;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.jbi.component.ComponentContext;
+import javax.jbi.messaging.DeliveryChannel;
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.MessageExchange;
+import javax.jbi.messaging.NormalizedMessage;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import javax.xml.namespace.QName;
+
+import org.apache.servicemix.JbiConstants;
+import org.apache.servicemix.common.ExchangeProcessor;
+import org.apache.servicemix.http.HttpEndpoint;
+import org.apache.servicemix.http.HttpLifeCycle;
+import org.apache.servicemix.http.HttpProcessor;
+import org.apache.servicemix.http.ServerManager;
+import org.apache.servicemix.soap.SoapFault;
+import org.apache.servicemix.soap.SoapHelper;
+import org.apache.servicemix.soap.handlers.AddressingInHandler;
+import org.apache.servicemix.soap.marshalers.JBIMarshaler;
+import org.apache.servicemix.soap.marshalers.SoapMarshaler;
+import org.apache.servicemix.soap.marshalers.SoapMessage;
+import org.mortbay.http.HttpContext;
+
+public class ConsumerProcessor implements ExchangeProcessor, HttpProcessor {
+
+ public static final URI IN_ONLY = URI.create("http://www.w3.org/2004/08/wsdl/in-only");
+ public static final URI IN_OUT = URI.create("http://www.w3.org/2004/08/wsdl/in-out");
+ public static final URI ROBUST_IN_ONLY = URI.create("http://www.w3.org/2004/08/wsdl/robust-in-only");
+
+ protected HttpEndpoint endpoint;
+ protected HttpContext httpContext;
+ protected ComponentContext context;
+ protected DeliveryChannel channel;
+ protected SoapMarshaler soapMarshaler;
+ protected JBIMarshaler jbiMarshaler;
+ protected SoapHelper soapHelper;
+
+ public ConsumerProcessor(HttpEndpoint endpoint) {
+ this.endpoint = endpoint;
+ this.soapMarshaler = new SoapMarshaler(endpoint.isSoap());
+ this.soapHelper = new SoapHelper(endpoint);
+ this.soapHelper.addPolicy(new AddressingInHandler());
+ this.jbiMarshaler = new JBIMarshaler();
+ }
+
+ public void process(MessageExchange exchange) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
+ public void start() throws Exception {
+ String url = endpoint.getLocationURI();
+ httpContext = getServerManager().createContext(url, this);
+ httpContext.start();
+ context = endpoint.getServiceUnit().getComponent().getComponentContext();
+ channel = context.getDeliveryChannel();
+ }
+
+ public void stop() throws Exception {
+ httpContext.stop();
+ getServerManager().remove(httpContext);
+ }
+
+ public void process(HttpServletRequest request, HttpServletResponse response) throws Exception {
+ QName envelopeName = null;
+ try {
+ SoapMessage message = soapMarshaler.createReader().read(request.getInputStream(),
+ request.getHeader("Content-Type"));
+ MessageExchange exchange = soapHelper.createExchange(message);
+ NormalizedMessage inMessage = exchange.getMessage("in");
+ inMessage.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(request));
+ boolean result = channel.sendSync(exchange);
+ if (!result) {
+ throw new Exception("Error sending exchange: aborted");
+ }
+ if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ exchange.setStatus(ExchangeStatus.DONE);
+ channel.send(exchange);
+ if (exchange.getError() != null) {
+ throw new Exception(exchange.getError());
+ } else if (exchange.getFault() != null) {
+ throw new SoapFault(SoapFault.RECEIVER, null, null, null, exchange.getFault().getContent());
+ } else {
+ throw new Exception("Unkown Error");
+ }
+ } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
+ NormalizedMessage outMsg = exchange.getMessage("out");
+ if (outMsg != null) {
+ SoapMessage out = new SoapMessage();
+ jbiMarshaler.fromNMS(out, outMsg);
+ soapMarshaler.createWriter(out).write(response.getOutputStream());
+ }
+ exchange.setStatus(ExchangeStatus.DONE);
+ channel.send(exchange);
+ }
+ } catch (SoapFault fault) {
+ if (SoapFault.SENDER.equals(fault.getCode())) {
+ response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
+ } else {
+ response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+ }
+ SoapMessage soapFault = new SoapMessage();
+ soapFault.setFault(fault);
+ soapFault.setEnvelopeName(envelopeName);
+ soapMarshaler.createWriter(soapFault).write(response.getOutputStream());
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ protected Map getHeaders(HttpServletRequest request) {
+ Map headers = new HashMap();
+ Enumeration enumeration = request.getHeaderNames();
+ while (enumeration.hasMoreElements()) {
+ String name = (String) enumeration.nextElement();
+ String value = request.getHeader(name);
+ headers.put(name, value);
+ }
+ return headers;
+ }
+
+ protected ServerManager getServerManager() {
+ HttpLifeCycle lf = (HttpLifeCycle) endpoint.getServiceUnit().getComponent().getLifeCycle();
+ return lf.getServer();
+ }
+
+}
Copied: incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ProviderProcessor.java (from r367101, incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/ProviderProcessor.java)
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ProviderProcessor.java?p2=incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ProviderProcessor.java&p1=incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/ProviderProcessor.java&r1=367101&r2=370186&rev=370186&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/ProviderProcessor.java (original)
+++ incubator/servicemix/trunk/servicemix-http/src/main/java/org/apache/servicemix/http/sm/ProviderProcessor.java Wed Jan 18 09:01:53 2006
@@ -13,33 +13,70 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.servicemix.http;
+package org.apache.servicemix.http.processors;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
import javax.jbi.messaging.DeliveryChannel;
import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.Fault;
+import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.InOptionalOut;
import javax.jbi.messaging.InOut;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.NormalizedMessage;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.commons.httpclient.Header;
import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.httpclient.HttpClient;
+import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.URI;
+import org.apache.commons.httpclient.URIException;
+import org.apache.commons.httpclient.methods.ByteArrayRequestEntity;
import org.apache.commons.httpclient.methods.PostMethod;
+import org.apache.commons.httpclient.methods.RequestEntity;
+import org.apache.servicemix.JbiConstants;
import org.apache.servicemix.common.ExchangeProcessor;
-
+import org.apache.servicemix.http.HttpEndpoint;
+import org.apache.servicemix.http.HttpLifeCycle;
+import org.apache.servicemix.soap.marshalers.JBIMarshaler;
+import org.apache.servicemix.soap.marshalers.SoapMarshaler;
+import org.apache.servicemix.soap.marshalers.SoapMessage;
+import org.apache.servicemix.soap.marshalers.SoapReader;
+import org.apache.servicemix.soap.marshalers.SoapWriter;
+
+/**
+ *
+ * @author Guillaume Nodet
+ * @version $Revision$
+ * @since 3.0
+ */
public class ProviderProcessor implements ExchangeProcessor {
protected HttpEndpoint endpoint;
protected HostConfiguration host;
- protected HttpClientMarshaler marshaler;
+ protected SoapMarshaler soapMarshaler;
+ protected JBIMarshaler jbiMarshaler;
protected DeliveryChannel channel;
- public ProviderProcessor(HttpEndpoint endpoint) throws Exception {
- this.endpoint = endpoint;
- this.host = new HostConfiguration();
- this.host.setHost(new URI(endpoint.getLocationURI(), false));
+ public ProviderProcessor(HttpEndpoint endpoint) {
+ try {
+ this.endpoint = endpoint;
+ this.soapMarshaler = new SoapMarshaler(endpoint.isSoap());
+ this.jbiMarshaler = new JBIMarshaler();
+ this.host = new HostConfiguration();
+ this.host.setHost(new URI(endpoint.getLocationURI(), false));
+ } catch (URIException e) {
+ throw new RuntimeException(e);
+ }
}
public void process(MessageExchange exchange) throws Exception {
@@ -51,14 +88,46 @@
return;
}
PostMethod method = new PostMethod(endpoint.getLocationURI());
- marshaler.fromNMS(method, exchange, exchange.getMessage("in"));
+ SoapMessage soapMessage = new SoapMessage();
+ NormalizedMessage nm = exchange.getMessage("in");
+ jbiMarshaler.fromNMS(soapMessage, nm);
+ SoapWriter writer = soapMarshaler.createWriter(soapMessage);
+ Map headers = (Map) nm.getProperty(JbiConstants.PROTOCOL_HEADERS);
+ if (headers != null) {
+ for (Iterator it = headers.keySet().iterator(); it.hasNext();) {
+ String name = (String) it.next();
+ String value = (String) headers.get(name);
+ method.addRequestHeader(name, value);
+ }
+ }
+ method.addRequestHeader("Content-Type", writer.getContentType());
+ method.setRequestEntity(writeMessage(writer));
int response = getClient().executeMethod(host, method);
if (response != HttpStatus.SC_OK) {
- throw new Exception("Invalid status response: " + response);
+ if (exchange instanceof InOnly == false) {
+ Fault fault = exchange.createFault();
+ SoapReader reader = soapMarshaler.createReader();
+ Header contentType = method.getResponseHeader("Content-Type");
+ soapMessage = reader.read(method.getResponseBodyAsStream(),
+ contentType != null ? contentType.getValue() : null);
+ fault.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
+ jbiMarshaler.toNMS(fault, soapMessage);
+ exchange.setFault(fault);
+ exchange.setStatus(ExchangeStatus.ERROR);
+ channel.send(exchange);
+ return;
+ } else {
+ throw new Exception("Invalid status response: " + response);
+ }
}
if (exchange instanceof InOut) {
NormalizedMessage msg = exchange.createMessage();
- marshaler.toNMS(msg, method);
+ SoapReader reader = soapMarshaler.createReader();
+ Header contentType = method.getResponseHeader("Content-Type");
+ soapMessage = reader.read(method.getResponseBodyAsStream(),
+ contentType != null ? contentType.getValue() : null);
+ msg.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
+ jbiMarshaler.toNMS(msg, soapMessage);
((InOut) exchange).setOutMessage(msg);
channel.sendSync(exchange);
} else if (exchange instanceof InOptionalOut) {
@@ -67,7 +136,11 @@
channel.send(exchange);
} else {
NormalizedMessage msg = exchange.createMessage();
- marshaler.toNMS(msg, method);
+ SoapReader reader = soapMarshaler.createReader();
+ soapMessage = reader.read(method.getResponseBodyAsStream(),
+ method.getResponseHeader("Content-Type").getValue());
+ msg.setProperty(JbiConstants.PROTOCOL_HEADERS, getHeaders(method));
+ jbiMarshaler.toNMS(msg, soapMessage);
((InOptionalOut) exchange).setOutMessage(msg);
channel.sendSync(exchange);
}
@@ -79,20 +152,75 @@
public void start() throws Exception {
channel = endpoint.getServiceUnit().getComponent().getComponentContext().getDeliveryChannel();
- HttpLifeCycle lf = (HttpLifeCycle) endpoint.getServiceUnit().getComponent().getLifeCycle();
- if (lf.getConfiguration().isStreamingEnabled()) {
- this.marshaler = new HttpStreamingClientMarshaler();
- } else {
- this.marshaler = new HttpClientMarshaler();
- }
}
public void stop() throws Exception {
}
+ protected Map getHeaders(HttpServletRequest request) {
+ Map headers = new HashMap();
+ Enumeration enumeration = request.getHeaderNames();
+ while (enumeration.hasMoreElements()) {
+ String name = (String) enumeration.nextElement();
+ String value = request.getHeader(name);
+ headers.put(name, value);
+ }
+ return headers;
+ }
+
+ protected Map getHeaders(HttpMethod method) {
+ Map headers = new HashMap();
+ Header[] h = method.getResponseHeaders();
+ for (int i = 0; i < h.length; i++) {
+ headers.put(h[i].getName(), h[i].getValue());
+ }
+ return headers;
+ }
+
+ protected RequestEntity writeMessage(SoapWriter writer) throws Exception {
+ HttpLifeCycle lf = (HttpLifeCycle) endpoint.getServiceUnit().getComponent().getLifeCycle();
+ if (lf.getConfiguration().isStreamingEnabled()) {
+ return new StreamingRequestEntity(writer);
+ } else {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ writer.write(baos);
+ return new ByteArrayRequestEntity(baos.toByteArray(), writer.getContentType());
+ }
+ }
+
protected HttpClient getClient() {
HttpLifeCycle lf = (HttpLifeCycle) endpoint.getServiceUnit().getComponent().getLifeCycle();
return lf.getClient();
}
+ public static class StreamingRequestEntity implements RequestEntity {
+
+ private SoapWriter writer;
+
+ public StreamingRequestEntity(SoapWriter writer) {
+ this.writer = writer;
+ }
+
+ public boolean isRepeatable() {
+ return false;
+ }
+
+ public void writeRequest(OutputStream out) throws IOException {
+ try {
+ writer.write(out);
+ } catch (Exception e) {
+ throw (IOException) new IOException("Could not write request").initCause(e);
+ }
+ }
+
+ public long getContentLength() {
+ // not known so we send negative value
+ return -1;
+ }
+
+ public String getContentType() {
+ return writer.getContentType();
+ }
+
+ }
}