You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@synapse.apache.org by as...@apache.org on 2007/01/22 11:34:38 UTC
svn commit: r498575 - in /webservices/synapse/trunk/java: ./ modules/core/
modules/core/src/test/java/org/apache/synapse/config/xml/
modules/core/src/test/java/org/apache/synapse/n2n/ modules/extensions/
modules/mar/ modules/nhttp/ modules/nhttp/src/ m...
Author: asankha
Date: Mon Jan 22 02:34:36 2007
New Revision: 498575
URL: http://svn.apache.org/viewvc?view=rev&rev=498575
Log:
checkin new NIO transport based on HttpCore and NIO extensions
fix build break and updated test case where an invalid name attribute was present
change version to SNAPSHOT
Added:
webservices/synapse/trunk/java/modules/nhttp/pom.xml
webservices/synapse/trunk/java/modules/nhttp/src/
webservices/synapse/trunk/java/modules/nhttp/src/org/
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Axis2HttpRequest.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientHandler.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientWorker.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOListener.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOSender.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerHandler.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerWorker.java
webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Util.java
Modified:
webservices/synapse/trunk/java/modules/core/pom.xml
webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/config/xml/SendMediatorSerializationTest.java
webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/n2n/SynapseCommodityServiceTest.java
webservices/synapse/trunk/java/modules/extensions/pom.xml
webservices/synapse/trunk/java/modules/mar/pom.xml
webservices/synapse/trunk/java/modules/samples/pom.xml
webservices/synapse/trunk/java/pom.xml
webservices/synapse/trunk/java/repository/conf/axis2.xml
Modified: webservices/synapse/trunk/java/modules/core/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/core/pom.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/core/pom.xml (original)
+++ webservices/synapse/trunk/java/modules/core/pom.xml Mon Jan 22 02:34:36 2007
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.synapse</groupId>
<artifactId>Apache-Synapse</artifactId>
- <version>0.91</version>
+ <version>SNAPSHOT</version>
</parent>
<groupId>org.apache.synapse</groupId>
@@ -121,4 +121,11 @@
</plugins>
</build>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.synapse</groupId>
+ <artifactId>synapse-nhttp</artifactId>
+ </dependency>
+ </dependencies>
</project>
Modified: webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/config/xml/SendMediatorSerializationTest.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/config/xml/SendMediatorSerializationTest.java?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/config/xml/SendMediatorSerializationTest.java (original)
+++ webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/config/xml/SendMediatorSerializationTest.java Mon Jan 22 02:34:36 2007
@@ -45,7 +45,7 @@
public void testSendMediatorWithSingleEndpoint() {
String sendConfiguration = "<syn:send xmlns:syn=\"http://ws.apache.org/ns/synapse\">" +
- "<syn:endpoint name=\"myendpoint\" address=\"http://ws.apache.org/axis2/services/Service1\" />" +
+ "<syn:endpoint address=\"http://ws.apache.org/axis2/services/Service1\" />" +
"</syn:send>";
try {
@@ -58,8 +58,8 @@
public void testSendMediatorWithMultipleEndpoints() {
String sendConfiguration = "<syn:send xmlns:syn=\"http://ws.apache.org/ns/synapse\">" +
- "<syn:endpoint name=\"myendpoint1\" address=\"http://ws.apache.org/axis2/services/Service1\" />" +
- "<syn:endpoint name=\"myendpoint2\" address=\"http://ws.apache.org/axis2/services/Service2\" />" +
+ "<syn:endpoint address=\"http://ws.apache.org/axis2/services/Service1\" />" +
+ "<syn:endpoint address=\"http://ws.apache.org/axis2/services/Service2\" />" +
"</syn:send>";
try {
Modified: webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/n2n/SynapseCommodityServiceTest.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/n2n/SynapseCommodityServiceTest.java?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/n2n/SynapseCommodityServiceTest.java (original)
+++ webservices/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/n2n/SynapseCommodityServiceTest.java Mon Jan 22 02:34:36 2007
@@ -30,18 +30,24 @@
import org.apache.axis2.context.ConfigurationContext;
import org.apache.axis2.context.ConfigurationContextFactory;
import org.apache.axis2.description.AxisService;
+import org.apache.axis2.description.TransportInDescription;
import org.apache.axis2.rpc.receivers.RPCMessageReceiver;
import org.apache.axis2.transport.http.SimpleHTTPServer;
+import org.apache.axis2.transport.nhttp.HttpCoreNIOListener;
+import org.apache.axis2.transport.TransportListener;
+import org.apache.axis2.engine.ListenerManager;
+import org.apache.axis2.AxisFault;
import org.apache.synapse.Constants;
import org.apache.synapse.utils.Services;
import javax.xml.namespace.QName;
+import java.util.Iterator;
public class SynapseCommodityServiceTest extends TestCase {
- private SimpleHTTPServer synapseServer = null;
- private SimpleHTTPServer businessServer = null;
+ private TransportListener synapseServer = null;
+ private TransportListener businessServer = null;
protected void setUp() throws java.lang.Exception {
// Initializing Synapse repository
@@ -68,17 +74,38 @@
"http://business.org", "http://business.org");
businessConfigCtx.getAxisConfiguration().addService(businessService);
- synapseServer = new SimpleHTTPServer(synapseConfigCtx, 10000);
- businessServer = new SimpleHTTPServer(businessConfigCtx, 10001);
-
- //starting servers
- synapseServer.start();
- businessServer.start();
+ TransportInDescription synTrsIn = (TransportInDescription)
+ synapseConfigCtx.getAxisConfiguration().getTransportsIn().get(new QName("http"));
+ synTrsIn.getParameter("port").setValue("10000");
+ startServer(synapseConfigCtx);
+
+ TransportInDescription busTrsIn = (TransportInDescription)
+ businessConfigCtx.getAxisConfiguration().getTransportsIn().get(new QName("http"));
+ busTrsIn.getParameter("port").setValue("10001");
+ startServer(businessConfigCtx);
+ System.out.println("");
}
protected void tearDown() throws java.lang.Exception {
- businessServer.stop();
- synapseServer.stop();
+ //businessServer.stop();
+ //synapseServer.stop();
+ }
+
+ private void startServer(ConfigurationContext configctx) throws AxisFault {
+ ListenerManager listenerManager = configctx.getListenerManager();
+ if (listenerManager == null) {
+ listenerManager = new ListenerManager();
+ listenerManager.init(configctx);
+ }
+
+ Iterator iter = configctx.getAxisConfiguration().
+ getTransportsIn().keySet().iterator();
+ while (iter.hasNext()) {
+ QName trp = (QName) iter.next();
+ TransportInDescription trsIn = (TransportInDescription)
+ configctx.getAxisConfiguration().getTransportsIn().get(trp);
+ listenerManager.addListener(trsIn, false);
+ }
}
public void testN2N() throws Exception {
Modified: webservices/synapse/trunk/java/modules/extensions/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/extensions/pom.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/extensions/pom.xml (original)
+++ webservices/synapse/trunk/java/modules/extensions/pom.xml Mon Jan 22 02:34:36 2007
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.synapse</groupId>
<artifactId>Apache-Synapse</artifactId>
- <version>0.91</version>
+ <version>SNAPSHOT</version>
</parent>
<groupId>org.apache.synapse</groupId>
Modified: webservices/synapse/trunk/java/modules/mar/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/mar/pom.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/mar/pom.xml (original)
+++ webservices/synapse/trunk/java/modules/mar/pom.xml Mon Jan 22 02:34:36 2007
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.synapse</groupId>
<artifactId>Apache-Synapse</artifactId>
- <version>0.91</version>
+ <version>SNAPSHOT</version>
</parent>
<groupId>org.apache.synapse</groupId>
Added: webservices/synapse/trunk/java/modules/nhttp/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/pom.xml?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/pom.xml (added)
+++ webservices/synapse/trunk/java/modules/nhttp/pom.xml Mon Jan 22 02:34:36 2007
@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you 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.
+ -->
+
+<project
+ xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.synapse</groupId>
+ <artifactId>Apache-Synapse</artifactId>
+ <version>SNAPSHOT</version>
+ </parent>
+
+ <artifactId>synapse-nhttp</artifactId>
+ <name>Apache Synapse - HttpCore-NIO</name>
+ <description>Apache Synapse - HttpCore-NIO</description>
+ <packaging>jar</packaging>
+
+ <build>
+ <sourceDirectory>src</sourceDirectory>
+ </build>
+
+ <dependencies>
+
+ </dependencies>
+
+ <properties>
+
+ </properties>
+
+</project>
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Axis2HttpRequest.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Axis2HttpRequest.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Axis2HttpRequest.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Axis2HttpRequest.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.AxisFault;
+import org.apache.http.*;
+import org.apache.http.entity.BasicHttpEntity;
+import org.apache.http.message.HttpPost;
+import org.apache.axiom.om.OMOutputFormat;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.xml.stream.XMLStreamException;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.channels.Pipe;
+import java.nio.channels.Channels;
+
+/**
+ * Represents an outgoing Axis2 HTTP/s request. It holds the EPR of the destination, the
+ * Axis2 MessageContext to be sent, an HttpHost object which captures information about the
+ * destination, and a Pipe used to write the message stream to the destination
+ */
+public class Axis2HttpRequest {
+
+ private static final Log log = LogFactory.getLog(Axis2HttpRequest.class);
+
+ /** the EPR of the destination */
+ private EndpointReference epr = null;
+ /** the HttpHost that contains the HTTP connection information */
+ private HttpHost httpHost = null;
+ /** the message context being sent */
+ private MessageContext msgContext = null;
+ /** the Pipe which facilitates the serialization output to be written to the channel */
+ private Pipe pipe = null;
+
+ public Axis2HttpRequest(EndpointReference epr, HttpHost httpHost, MessageContext msgContext) {
+ this.epr = epr;
+ this.httpHost = httpHost;
+ this.msgContext = msgContext;
+ try {
+ this.pipe = Pipe.open();
+ } catch (IOException e) {
+ log.error("Error creating pipe to write message body");
+ }
+ }
+
+ public EndpointReference getEpr() {
+ return epr;
+ }
+
+ public HttpHost getHttpHost() {
+ return httpHost;
+ }
+
+ public MessageContext getMsgContext() {
+ return msgContext;
+ }
+
+ /**
+ * Create and return a new HttpPost request to the destination EPR
+ * @return the HttpRequest to be sent out
+ */
+ public HttpRequest getRequest() {
+ HttpPost httpRequest = new HttpPost(epr.getAddress());
+ httpRequest.setEntity(new BasicHttpEntity());
+ return httpRequest;
+ }
+
+ /**
+ * Return the source channel of the pipe that bridges the serialized output to the socket
+ * @return source channel to read serialized message contents
+ */
+ public Pipe.SourceChannel getSourceChannel() {
+ log.debug("get source channel of the pipe on which the outgoing response is written");
+ return pipe.source();
+ }
+
+ /**
+ * Start streaming the message into the Pipe, so that the contents could be read off the source
+ * channel returned by getSourceChannel()
+ * @throws AxisFault on error
+ */
+ public void streamMessageContents() throws AxisFault {
+
+ log.debug("start streaming outgoing http request");
+ OutputStream out = Channels.newOutputStream(pipe.sink());
+ OMOutputFormat format = Util.getOMOutputFormat(msgContext);
+
+ try {
+ (msgContext.isDoingREST() ?
+ msgContext.getEnvelope().getBody().getFirstElement() : msgContext.getEnvelope())
+ .serializeAndConsume(out, format);
+ } catch (XMLStreamException e) {
+ handleException("Error serializing response message", e);
+ }
+
+ try {
+ out.flush();
+ out.close();
+ } catch (IOException e) {
+ handleException("Error closing outgoing message stream", e);
+ }
+ }
+
+ // -------------- utility methods -------------
+ private void handleException(String msg, Exception e) throws AxisFault {
+ log.error(msg, e);
+ throw new AxisFault(msg, e);
+ }
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientHandler.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientHandler.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientHandler.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientHandler.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,283 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import org.apache.http.nio.NHttpClientHandler;
+import org.apache.http.nio.NHttpClientConnection;
+import org.apache.http.nio.ContentDecoder;
+import org.apache.http.nio.ContentEncoder;
+import org.apache.http.params.HttpParams;
+import org.apache.http.*;
+import org.apache.http.entity.BasicHttpEntity;
+import org.apache.http.impl.DefaultConnectionReuseStrategy;
+import org.apache.http.protocol.*;
+import org.apache.axis2.util.threadpool.DefaultThreadFactory;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.MessageContext;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.Pipe;
+import java.io.IOException;
+
+import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+import edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor;
+import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
+import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
+
+/**
+ * The client connection handler. An instance of this class is used by each IOReactor, to
+ * process every connection. Hence this class should not store any data related to a single
+ * connection - as this is being shared.
+ */
+public class ClientHandler implements NHttpClientHandler {
+
+ private static final Log log = LogFactory.getLog(ClientHandler.class);
+
+ /** the HTTP protocol parameters to adhere to for outgoing messages */
+ private final HttpParams params;
+ /** the HttpProcessor for response messages received */
+ private final HttpProcessor httpProcessor;
+ /** the connection re-use strategy */
+ private final ConnectionReuseStrategy connStrategy;
+
+ /** the Axis2 configuration context */
+ ConfigurationContext cfgCtx = null;
+
+ private Executor workerPool = null;
+ private static final int WORKERS_MAX_THREADS = 40;
+ private static final long WORKER_KEEP_ALIVE = 100L;
+
+ private static final String REQUEST_BUFFER = "request-buffer";
+ private static final String RESPONSE_BUFFER = "response-buffer";
+ private static final String OUTGOING_MESSAGE_CONTEXT = "axis2_message_context";
+ private static final String REQUEST_SOURCE_CHANNEL = "request-source-channel";
+ private static final String RESPONSE_SINK_CHANNEL = "request-sink-channel";
+
+ /**
+ * Create an instance of this client connection handler using the Axis2 configuration
+ * context and Http protocol parameters given
+ * @param cfgCtx the Axis2 configuration context
+ * @param params the Http protocol parameters to adhere to
+ */
+ public ClientHandler(final ConfigurationContext cfgCtx, final HttpParams params) {
+ super();
+ this.cfgCtx = cfgCtx;
+ this.params = params;
+ this.httpProcessor = getHttpProcessor();
+ this.connStrategy = new DefaultConnectionReuseStrategy();
+
+ workerPool = new ThreadPoolExecutor(
+ 1, WORKERS_MAX_THREADS, WORKER_KEEP_ALIVE, TimeUnit.SECONDS,
+ new LinkedBlockingQueue(),
+ new DefaultThreadFactory(new ThreadGroup("Client Worker thread group"), "HttpClientWorker"));
+ }
+
+ /**
+ * Invoked when the destination is connected
+ * @param conn the connection being processed
+ * @param attachment the attachment set previously
+ */
+ public void connected(final NHttpClientConnection conn, final Object attachment) {
+ try {
+ HttpContext context = conn.getContext();
+ Axis2HttpRequest axis2Req = (Axis2HttpRequest) attachment;
+
+ context.setAttribute(HttpExecutionContext.HTTP_CONNECTION, conn);
+ context.setAttribute(HttpExecutionContext.HTTP_TARGET_HOST, axis2Req.getHttpHost());
+
+ // allocate temporary buffers to process this request
+ context.setAttribute(REQUEST_BUFFER, ByteBuffer.allocate(2048));
+ context.setAttribute(RESPONSE_BUFFER, ByteBuffer.allocate(2048));
+
+ context.setAttribute(OUTGOING_MESSAGE_CONTEXT, axis2Req.getMsgContext());
+ context.setAttribute(REQUEST_SOURCE_CHANNEL, axis2Req.getSourceChannel());
+
+ HttpRequest request = axis2Req.getRequest();
+ request.getParams().setDefaults(this.params);
+ this.httpProcessor.process(request, context);
+
+ conn.submitRequest(request);
+ context.setAttribute(HttpExecutionContext.HTTP_REQUEST, request);
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ } catch (HttpException e) {
+ handleException("Unexpected HTTP protocol error: " + e.getMessage(), e, conn);
+ }
+ }
+
+ public void closed(final NHttpClientConnection conn) {
+ log.trace("Connection closed");
+ }
+
+ /**
+ * Handle connection timeouts by shutting down the connections
+ * @param conn the connection being processed
+ */
+ public void timeout(final NHttpClientConnection conn) {
+ log.debug("Connection Timeout");
+ shutdownConnection(conn);
+ }
+
+ /**
+ * Handle Http protocol violations encountered while reading from underlying channels
+ * @param conn the connection being processed
+ * @param e the exception encountered
+ */
+ public void exception(final NHttpClientConnection conn, final HttpException e) {
+ log.error("HTTP protocol violation : " + e.getMessage());
+ shutdownConnection(conn);
+ }
+
+ /**
+ * Handle IO errors while reading or writing to underlying channels
+ * @param conn the connection being processed
+ * @param e the exception encountered
+ */
+ public void exception(final NHttpClientConnection conn, final IOException e) {
+ log.error("I/O error : " + e.getMessage());
+ shutdownConnection(conn);
+ }
+
+ /**
+ * Process ready input (i.e. response from remote server)
+ * @param conn connection being processed
+ * @param decoder the content decoder in use
+ */
+ public void inputReady(final NHttpClientConnection conn, final ContentDecoder decoder) {
+ HttpContext context = conn.getContext();
+ HttpResponse response = conn.getHttpResponse();
+ Pipe.SinkChannel sink = (Pipe.SinkChannel) context.getAttribute(RESPONSE_SINK_CHANNEL);
+ ByteBuffer inbuf = (ByteBuffer) context.getAttribute(REQUEST_BUFFER);
+
+ try {
+ while (decoder.read(inbuf) > 0) {
+ inbuf.flip();
+ sink.write(inbuf);
+ inbuf.compact();
+ }
+
+ if (decoder.isCompleted()) {
+ sink.close();
+ if (!connStrategy.keepAlive(response, context)) {
+ conn.close();
+ }
+ }
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ }
+ }
+
+ /**
+ * Process ready output (i.e. write request to remote server)
+ * @param conn the connection being processed
+ * @param encoder the encoder in use
+ */
+ public void outputReady(final NHttpClientConnection conn, final ContentEncoder encoder) {
+ HttpContext context = conn.getContext();
+ HttpResponse response = conn.getHttpResponse();
+
+ Pipe.SourceChannel source = (Pipe.SourceChannel) context.getAttribute(REQUEST_SOURCE_CHANNEL);
+ ByteBuffer outbuf = (ByteBuffer) context.getAttribute(RESPONSE_BUFFER);
+
+ try {
+ int bytesRead = source.read(outbuf);
+ if (bytesRead == -1) {
+ encoder.complete();
+ } else {
+ outbuf.flip();
+ encoder.write(outbuf);
+ outbuf.compact();
+ }
+
+ if (encoder.isCompleted()) {
+ source.close();
+ }
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ }
+ }
+
+ /**
+ * Process a response received for the request sent out
+ * @param conn the connection being processed
+ */
+ public void responseReceived(final NHttpClientConnection conn) {
+ HttpContext context = conn.getContext();
+ HttpResponse response = conn.getHttpResponse();
+
+ try {
+ Pipe responsePipe = Pipe.open();
+ context.setAttribute(RESPONSE_SINK_CHANNEL, responsePipe.sink());
+
+ BasicHttpEntity entity = new BasicHttpEntity();
+ if (response.getStatusLine().getHttpVersion().greaterEquals(HttpVersion.HTTP_1_1)) {
+ entity.setChunked(true);
+ }
+ response.setEntity(entity);
+ context.setAttribute(HttpContext.HTTP_RESPONSE, response);
+
+ workerPool.execute(
+ new ClientWorker(cfgCtx, Channels.newInputStream(responsePipe.source()),
+ (MessageContext) context.getAttribute(OUTGOING_MESSAGE_CONTEXT)));
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ }
+ }
+
+ // ----------- utility methods -----------
+
+ private void handleException(String msg, Exception e, NHttpClientConnection conn) {
+ log.error(msg, e);
+ if (conn != null) {
+ shutdownConnection(conn);
+ }
+ }
+
+ /**
+ * Shutdown the connection ignoring any IO errors during the process
+ * @param conn the connection to be shutdown
+ */
+ private void shutdownConnection(final HttpConnection conn) {
+ try {
+ conn.shutdown();
+ } catch (IOException ignore) {}
+ }
+
+ /**
+ * Return the HttpProcessor for requests
+ * @return the HttpProcessor that processes requests
+ */
+ private HttpProcessor getHttpProcessor() {
+ BasicHttpProcessor httpProcessor = new BasicHttpProcessor();
+ httpProcessor.addInterceptor(new RequestContent());
+ httpProcessor.addInterceptor(new RequestTargetHost());
+ httpProcessor.addInterceptor(new RequestConnControl());
+ httpProcessor.addInterceptor(new RequestUserAgent());
+ httpProcessor.addInterceptor(new RequestExpectContinue());
+ return httpProcessor;
+ }
+
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientWorker.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientWorker.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientWorker.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ClientWorker.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.description.WSDL2Constants;
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.engine.AxisEngine;
+import org.apache.axis2.transport.TransportUtils;
+import org.apache.axiom.soap.SOAPEnvelope;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.InputStream;
+
+/**
+ * Performs processing of the HTTP response received for our outgoing request. An instance of this
+ * class is created to process each unique response.
+ */
+public class ClientWorker implements Runnable {
+
+ private static final Log log = LogFactory.getLog(ClientWorker.class);
+
+ /** the Axis2 configuration context */
+ private ConfigurationContext cfgCtx = null;
+ /** the response message context that would be created */
+ private MessageContext responseMsgCtx = null;
+ /** the InputStream out of which the response body should be read */
+ private InputStream in = null;
+ /** the original request message context */
+ private MessageContext outMsgCtx = null;
+
+ /**
+ * Create the thread that would process the response message received for the outgoing message
+ * context sent
+ * @param cfgCtx the Axis2 configuration context
+ * @param in the InputStream to read the body of the response message received
+ * @param outMsgCtx the original outgoing message context (i.e. corresponding request)
+ */
+ public ClientWorker(ConfigurationContext cfgCtx, InputStream in, MessageContext outMsgCtx) {
+
+ this.cfgCtx = cfgCtx;
+ this.in = in;
+ this.outMsgCtx = outMsgCtx;
+
+ try {
+ responseMsgCtx = outMsgCtx.getOperationContext().
+ getMessageContext(WSDL2Constants.MESSAGE_LABEL_IN);
+ } catch (AxisFault af) {
+ log.error("Error getting IN message context from the operation context", af);
+ return;
+ }
+
+ if (responseMsgCtx == null) {
+ log.error("Error getting IN message context from the operation context");
+ } else {
+ responseMsgCtx.setServerSide(true);
+ responseMsgCtx.setDoingREST(outMsgCtx.isDoingREST());
+ responseMsgCtx.setProperty(MessageContext.TRANSPORT_IN, outMsgCtx
+ .getProperty(MessageContext.TRANSPORT_IN));
+ responseMsgCtx.setTransportIn(outMsgCtx.getTransportIn());
+ responseMsgCtx.setTransportOut(outMsgCtx.getTransportOut());
+
+ responseMsgCtx.setOperationContext(outMsgCtx.getOperationContext());
+ responseMsgCtx.setConfigurationContext(outMsgCtx.getConfigurationContext());
+ //responseMsgCtx.getOptions().setRelationships(
+ // new RelatesTo[] { new RelatesTo(outMsgCtx.getMessageID()) });
+ responseMsgCtx.setTo(null);
+ }
+ }
+
+ /**
+ * Process the received response through Axis2
+ */
+ public void run() {
+ SOAPEnvelope envelope = null;
+ try {
+ envelope = TransportUtils.createSOAPMessage(
+ responseMsgCtx,
+ in,
+ outMsgCtx.getEnvelope().getNamespace().getNamespaceURI());
+ responseMsgCtx.setEnvelope(envelope);
+ } catch (AxisFault af) {
+ log.error("Fault creating response SOAP envelope", af);
+ return;
+ }
+
+ AxisEngine engine = new AxisEngine(cfgCtx);
+ try {
+ if (envelope.getBody().hasFault()) {
+ engine.receiveFault(responseMsgCtx);
+ } else {
+ engine.receive(responseMsgCtx);
+ }
+ } catch (AxisFault af) {
+ log.error("Fault processing response message through Axis2", af);
+ }
+ }
+
+ // -------------- utility methods -------------
+ private void handleException(String msg, Exception e) throws AxisFault {
+ log.error(msg, e);
+ throw new AxisFault(msg, e);
+ }
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOListener.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOListener.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOListener.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOListener.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,195 @@
+/*
+* Copyright 2004,2005 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.axis2.transport.nhttp;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.description.Parameter;
+import org.apache.axis2.description.TransportInDescription;
+import org.apache.axis2.transport.TransportListener;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.nio.reactor.IOEventDispatch;
+import org.apache.http.nio.reactor.ListeningIOReactor;
+import org.apache.http.nio.impl.DefaultServerIOEventDispatch;
+import org.apache.http.nio.impl.reactor.DefaultListeningIOReactor;
+import org.apache.http.nio.NHttpServiceHandler;
+import org.apache.http.impl.DefaultHttpParams;
+
+import java.io.InterruptedIOException;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+
+/**
+ * NIO transport listener for Axis2 based on HttpCore and NIO extensions
+ */
+public class HttpCoreNIOListener implements TransportListener {
+
+ private static final Log log = LogFactory.getLog(HttpCoreNIOListener.class);
+
+ /** The Axis2 configuration context */
+ private ConfigurationContext cfgCtx;
+ /** The IOReactor */
+ private ListeningIOReactor ioReactor = null;
+
+ /** The EPR prefix for services available over this transport */
+ private String serviceEPRPrefix;
+ /** The port to listen on, defaults to 8080 */
+ private int port = 8080;
+ /** The hostname to use, defaults to localhost */
+ private String host = "localhost";
+
+ /**
+ * configure and start the IO reactor on the specified port
+ * @param port port to start the listener on
+ */
+ private void startServerEngine(int port) {
+ HttpParams params = getServerParameters();
+ try {
+ ioReactor = new DefaultListeningIOReactor(2, params);
+ } catch (IOException e) {
+ log.error("Error starting the IOReactor", e);
+ }
+
+ NHttpServiceHandler handler = new ServerHandler(cfgCtx, params);
+ IOEventDispatch ioEventDispatch = new DefaultServerIOEventDispatch(handler, params);
+
+ try {
+ ioReactor.listen(new InetSocketAddress(port));
+ ioReactor.execute(ioEventDispatch);
+ } catch (InterruptedIOException ex) {
+ log.fatal("Reactor Interrupted");
+ } catch (IOException e) {
+ log.fatal("Encountered an I/O error: " + e.getMessage(), e);
+ }
+ log.info("Listener Shutdown");
+ }
+
+ /**
+ * get HTTP protocol parameters to which the listener must adhere to
+ * @return the applicable HTTP protocol parameters
+ */
+ private HttpParams getServerParameters() {
+ HttpParams params = new DefaultHttpParams(null);
+ params
+ .setIntParameter(HttpConnectionParams.SO_TIMEOUT, 30000)
+ .setIntParameter(HttpConnectionParams.SOCKET_BUFFER_SIZE, 8 * 1024)
+ .setBooleanParameter(HttpConnectionParams.STALE_CONNECTION_CHECK, false)
+ .setBooleanParameter(HttpConnectionParams.TCP_NODELAY, true)
+ .setParameter(HttpProtocolParams.ORIGIN_SERVER, "Synapse-HttpComponents-NIO");
+ return params;
+ }
+
+ /**
+ * Initialize the transport listener, and execute reactor in new seperate thread
+ * @param cfgCtx the Axis2 configuration context
+ * @param transprtIn the description of the http/s transport from Axis2 configuration
+ * @throws AxisFault on error
+ */
+ public void init(ConfigurationContext cfgCtx, TransportInDescription transprtIn) throws AxisFault {
+
+ this.cfgCtx = cfgCtx;
+ Parameter param = transprtIn.getParameter(PARAM_PORT);
+ if (param != null) {
+ port = Integer.parseInt((String) param.getValue());
+ }
+
+ param = transprtIn.getParameter(HOST_ADDRESS);
+ if (param != null) {
+ host = ((String) param.getValue()).trim();
+ } else {
+ try {
+ host = java.net.InetAddress.getLocalHost().getHostName();
+ } catch (UnknownHostException e) {
+ log.warn("Unable to lookup local host name, using 'localhost'");
+ }
+ }
+
+ serviceEPRPrefix = "http://" + host + (port == 80 ? "" : ":" + port) +
+ cfgCtx.getServiceContextPath() + "/";
+ }
+
+ /**
+ * Start the transport listener on a new thread
+ * @throws AxisFault
+ */
+ public void start() throws AxisFault {
+ log.debug("Starting Listener...");
+ // start the Listener in a new seperate thread
+ Thread t = new Thread(new Runnable() {
+ public void run() {
+ try {
+ startServerEngine(port);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }, "HttpCoreNIOListener");
+
+ t.start();
+ log.info("Listener started, accepting connections...");
+ }
+
+ /**
+ * Stop the listener
+ * @throws AxisFault on error
+ */
+ public void stop() throws AxisFault {
+ try {
+ ioReactor.shutdown();
+ log.info("Listener shut down");
+ } catch (IOException e) {
+ handleException("Error shutting down IOReactor", e);
+ }
+ }
+
+ /**
+ * Return the EPR for the given service (implements deprecated method temporarily)
+ */
+ public EndpointReference getEPRForService(String serviceName, String ip) throws AxisFault {
+ return new EndpointReference(serviceEPRPrefix + serviceName);
+ }
+
+ /**
+ * Return the EPRs for the given service over this transport
+ * @param serviceName name of the service
+ * @param ip IP address
+ * @return the EndpointReferences for this service over the transport
+ * @throws AxisFault on error
+ */
+ public EndpointReference[] getEPRsForService(String serviceName, String ip) throws AxisFault {
+ EndpointReference[] endpointReferences = new EndpointReference[1];
+ endpointReferences[0] = new EndpointReference(serviceEPRPrefix + serviceName);
+ return endpointReferences;
+ }
+
+ // -------------- utility methods -------------
+ private void handleException(String msg, Exception e) throws AxisFault {
+ log.error(msg, e);
+ throw new AxisFault(msg, e);
+ }
+
+ private void handleException(String msg) throws AxisFault {
+ log.error(msg);
+ throw new AxisFault(msg);
+ }
+
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOSender.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOSender.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOSender.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/HttpCoreNIOSender.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,265 @@
+/*
+* Copyright 2004,2005 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.axis2.transport.nhttp;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.Constants;
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.addressing.AddressingConstants;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.description.TransportOutDescription;
+import org.apache.axis2.handlers.AbstractHandler;
+import org.apache.axis2.transport.TransportSender;
+import org.apache.axis2.transport.OutTransportInfo;
+import org.apache.axiom.om.OMOutputFormat;
+import org.apache.http.nio.NHttpClientHandler;
+import org.apache.http.nio.impl.reactor.DefaultConnectingIOReactor;
+import org.apache.http.nio.impl.DefaultClientIOEventDispatch;
+import org.apache.http.nio.reactor.ConnectingIOReactor;
+import org.apache.http.nio.reactor.IOEventDispatch;
+import org.apache.http.nio.reactor.SessionRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpHost;
+import org.apache.http.impl.DefaultHttpParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpProtocolParams;
+import org.apache.http.protocol.HTTP;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.xml.stream.XMLStreamException;
+import java.io.OutputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.InetSocketAddress;
+
+/**
+ * NIO transport sender for Axis2 based on HttpCore and NIO extensions
+ */
+public class HttpCoreNIOSender extends AbstractHandler implements TransportSender {
+
+ private static final Log log = LogFactory.getLog(HttpCoreNIOSender.class);
+
+ /** The Axis2 configuration context */
+ private ConfigurationContext cfgCtx;
+ /** The IOReactor */
+ private ConnectingIOReactor ioReactor = null;
+
+ /**
+ * Initialize the transport sender, and execute reactor in new seperate thread
+ * @param cfgCtx the Axis2 configuration context
+ * @param transportOut the description of the http/s transport from Axis2 configuration
+ * @throws AxisFault thrown on an error
+ */
+ public void init(ConfigurationContext cfgCtx, TransportOutDescription transportOut) throws AxisFault {
+ this.cfgCtx = cfgCtx;
+
+ // start the Sender in a new seperate thread
+ Thread t = new Thread(new Runnable() {
+ public void run() {
+ executeClientEngine();
+ }
+ }, "HttpCoreNIOSender");
+ t.start();
+ log.info("Sender started");
+ }
+
+ /**
+ * Configure and start the IOReactor
+ */
+ private void executeClientEngine() {
+
+ HttpParams params = getClientParameters();
+ try {
+ ioReactor = new DefaultConnectingIOReactor(2, params);
+ } catch (IOException e) {
+ log.error("Error starting the IOReactor", e);
+ }
+
+ NHttpClientHandler handler = new ClientHandler(cfgCtx, params);
+ IOEventDispatch ioEventDispatch = new DefaultClientIOEventDispatch(handler, params);
+
+ try {
+ ioReactor.execute(ioEventDispatch);
+ } catch (InterruptedIOException ex) {
+ log.fatal("Reactor Interrupted");
+ } catch (IOException e) {
+ log.fatal("Encountered an I/O error: " + e.getMessage(), e);
+ }
+ log.info("Sender Shutdown");
+ }
+
+ /**
+ * get HTTP protocol parameters to which the sender must adhere to
+ * @return the applicable HTTP protocol parameters
+ */
+ private HttpParams getClientParameters() {
+ HttpParams params = new DefaultHttpParams(null);
+ params
+ .setIntParameter(HttpConnectionParams.SO_TIMEOUT, 5000)
+ .setIntParameter(HttpConnectionParams.CONNECTION_TIMEOUT, 10000)
+ .setIntParameter(HttpConnectionParams.SOCKET_BUFFER_SIZE, 8 * 1024)
+ .setBooleanParameter(HttpConnectionParams.STALE_CONNECTION_CHECK, false)
+ .setBooleanParameter(HttpConnectionParams.TCP_NODELAY, true)
+ .setParameter(HttpProtocolParams.USER_AGENT, "Synapse-HttpComponents-NIO");
+ return params;
+ }
+
+ /**
+ * transport sender invocation from Axis2 core
+ * @param msgContext message to be sent
+ * @return the invocation response (always InvocationResponse.CONTINUE)
+ * @throws AxisFault on error
+ */
+ public InvocationResponse invoke(MessageContext msgContext) throws AxisFault {
+
+ EndpointReference epr = Util.getDestinationEPR(msgContext);
+ if (epr != null) {
+ if (!AddressingConstants.Final.WSA_NONE_URI.equals(epr.getAddress())) {
+ sendAsyncRequest(epr, msgContext);
+ } else {
+ handleException("Cannot send message to " + AddressingConstants.Final.WSA_NONE_URI);
+ }
+ } else {
+ if (msgContext.getProperty(Constants.OUT_TRANSPORT_INFO) != null) {
+ if (msgContext.getProperty(Constants.OUT_TRANSPORT_INFO) instanceof ServerWorker) {
+ sendAsyncResponse(msgContext);
+ } else {
+ sendUsingOutputStream(msgContext);
+ }
+ } else {
+ handleException("No valid destination EPR or OutputStream to send message");
+ }
+ }
+
+ if (msgContext.getOperationContext() != null) {
+ msgContext.getOperationContext().setProperty(
+ Constants.RESPONSE_WRITTEN, Constants.VALUE_TRUE);
+ }
+
+ return InvocationResponse.CONTINUE;
+ }
+
+ /**
+ * Send the request message asynchronously to the given EPR
+ * @param epr the destination EPR for the message
+ * @param msgContext the message being sent
+ * @throws AxisFault on error
+ */
+ private void sendAsyncRequest(EndpointReference epr, MessageContext msgContext) throws AxisFault {
+ try {
+ URL url = new URL(epr.getAddress());
+ HttpHost httpHost = new HttpHost(url.getHost(), url.getPort(), url.getProtocol());
+
+ Axis2HttpRequest axis2Req = new Axis2HttpRequest(epr, httpHost, msgContext);
+ SessionRequest req = ioReactor.connect(
+ new InetSocketAddress(url.getHost(), url.getPort()), null, axis2Req);
+
+ axis2Req.streamMessageContents();
+
+ } catch (MalformedURLException e) {
+ handleException("Malformed destination EPR : " + epr.getAddress(), e);
+ } catch (IOException e) {
+ handleException("IO Error while submiting request message for sending", e);
+ }
+ }
+
+ /**
+ * Send the passed in response message, asynchronously
+ * @param msgContext the message context to be sent
+ * @throws AxisFault on error
+ */
+ private void sendAsyncResponse(MessageContext msgContext) throws AxisFault {
+
+ ServerWorker worker = (ServerWorker) msgContext.getProperty(Constants.OUT_TRANSPORT_INFO);
+ HttpResponse response = worker.getResponse();
+
+ OMOutputFormat format = Util.getOMOutputFormat(msgContext);
+
+ response.setHeader(HTTP.CONTENT_TYPE, Util.getContentType(msgContext) + "; charset=" + format.getCharSetEncoding());
+
+ worker.getServiceHandler().commitResponse(worker.getConn(), response);
+
+ OutputStream out = worker.getOutputStream();
+ format.setDoOptimize(msgContext.isDoingMTOM());
+ try {
+ (msgContext.isDoingREST() ?
+ msgContext.getEnvelope().getBody().getFirstElement() : msgContext.getEnvelope())
+ .serializeAndConsume(out, format);
+ out.close();
+ } catch (XMLStreamException e) {
+ handleException("Error serializing response message", e);
+ } catch (IOException e) {
+ handleException("IO Error sending response message", e);
+ }
+ }
+
+ private void sendUsingOutputStream(MessageContext msgContext) throws AxisFault {
+ OMOutputFormat format = Util.getOMOutputFormat(msgContext);
+ OutputStream out =
+ (OutputStream) msgContext
+ .getProperty(MessageContext.TRANSPORT_OUT);
+
+ if (msgContext.isServerSide()) {
+ OutTransportInfo transportInfo =
+ (OutTransportInfo) msgContext.getProperty(Constants.OUT_TRANSPORT_INFO);
+
+ if (transportInfo != null) {
+ String encoding = Util.getContentType(msgContext) + "; charset=" + format.getCharSetEncoding();
+ transportInfo.setContentType(encoding);
+ } else {
+ throw new AxisFault(Constants.OUT_TRANSPORT_INFO + " has not been set");
+ }
+ }
+
+ try {
+ (msgContext.isDoingREST() ?
+ msgContext.getEnvelope().getBody().getFirstElement() : msgContext.getEnvelope())
+ .serializeAndConsume(out, format);
+ } catch (XMLStreamException e) {
+ handleException("Error serializing response message", e);
+ }
+ }
+
+
+ public void cleanup(MessageContext msgContext) throws AxisFault {
+ // do nothing
+ }
+
+ public void stop() {
+ try {
+ ioReactor.shutdown();
+ log.info("Sender shut down");
+ } catch (IOException e) {
+ log.warn("Error shutting down IOReactor", e);
+ }
+ }
+
+ // -------------- utility methods -------------
+ private void handleException(String msg, Exception e) throws AxisFault {
+ log.error(msg, e);
+ throw new AxisFault(msg, e);
+ }
+
+ private void handleException(String msg) throws AxisFault {
+ log.error(msg);
+ throw new AxisFault(msg);
+ }
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerHandler.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerHandler.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerHandler.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerHandler.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,291 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
+import edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor;
+import edu.emory.mathcs.backport.java.util.concurrent.TimeUnit;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.util.threadpool.DefaultThreadFactory;
+import org.apache.http.*;
+import org.apache.http.entity.BasicHttpEntity;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.impl.DefaultConnectionReuseStrategy;
+import org.apache.http.impl.DefaultHttpResponseFactory;
+import org.apache.http.nio.ContentDecoder;
+import org.apache.http.nio.ContentEncoder;
+import org.apache.http.nio.NHttpServerConnection;
+import org.apache.http.nio.NHttpServiceHandler;
+import org.apache.http.params.HttpParams;
+import org.apache.http.protocol.*;
+import org.apache.http.util.EncodingUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.Channels;
+import java.nio.channels.Pipe;
+
+/**
+ * The server connection handler. An instance of this class is used by each IOReactor, to
+ * process every connection. Hence this class should not store any data related to a single
+ * connection - as this is being shared.
+ */
+public class ServerHandler implements NHttpServiceHandler {
+
+ private static final Log log = LogFactory.getLog(ServerHandler.class);
+
+ /** the HTTP protocol parameters to adhere to */
+ private final HttpParams params;
+ /** the factory to create HTTP responses */
+ private final HttpResponseFactory responseFactory;
+ /** the HTTP response processor */
+ private final HttpProcessor httpProcessor;
+ /** the strategy to re-use connections */
+ private final ConnectionReuseStrategy connStrategy;
+
+ /** the Axis2 configuration context */
+ ConfigurationContext cfgCtx = null;
+
+ /** the thread pool to process requests */
+ private Executor workerPool = null;
+ private static final int WORKERS_MAX_THREADS = 40;
+ private static final long WORKER_KEEP_ALIVE = 100L;
+
+ private static final String REQUEST_SINK_CHANNEL = "request-sink-channel";
+ private static final String RESPONSE_SOURCE_CHANNEL = "response-source-channel";
+ private static final String REQUEST_BUFFER = "request-buffer";
+ private static final String RESPONSE_BUFFER = "response-buffer";
+
+ public ServerHandler(final ConfigurationContext cfgCtx, final HttpParams params) {
+ super();
+ this.cfgCtx = cfgCtx;
+ this.params = params;
+ this.responseFactory = new DefaultHttpResponseFactory();
+ this.httpProcessor = getHttpProcessor();
+ this.connStrategy = new DefaultConnectionReuseStrategy();
+
+ this.workerPool = new ThreadPoolExecutor(
+ 1, WORKERS_MAX_THREADS, WORKER_KEEP_ALIVE, TimeUnit.SECONDS,
+ new LinkedBlockingQueue(),
+ new DefaultThreadFactory(new ThreadGroup("Server Worker thread group"), "HttpServerWorker"));
+ }
+
+ /**
+ * Process a new incoming request
+ * @param conn the connection
+ */
+ public void requestReceived(final NHttpServerConnection conn) {
+
+ HttpContext context = conn.getContext();
+ HttpRequest request = conn.getHttpRequest();
+ context.setAttribute(HttpContext.HTTP_REQUEST, request);
+
+ // allocate temporary buffers to process this request
+ context.setAttribute(REQUEST_BUFFER, ByteBuffer.allocate(2048));
+ context.setAttribute(RESPONSE_BUFFER, ByteBuffer.allocate(2048));
+
+ try {
+ Pipe requestPipe = Pipe.open(); // the pipe used to process the request
+ Pipe responsePipe = Pipe.open(); // the pipe used to process the response
+ context.setAttribute(REQUEST_SINK_CHANNEL, requestPipe.sink());
+ context.setAttribute(RESPONSE_SOURCE_CHANNEL, responsePipe.source());
+
+ // create the default response to this request
+ HttpVersion httpVersion = request.getRequestLine().getHttpVersion();
+ HttpResponse response = responseFactory.newHttpResponse(httpVersion, HttpStatus.SC_OK);
+ response.setParams(this.params);
+
+ // create a basic HttpEntity using the source channel of the response pipe
+ BasicHttpEntity entity = new BasicHttpEntity();
+ entity.setContent(Channels.newInputStream(responsePipe.source()));
+ if (httpVersion.greaterEquals(HttpVersion.HTTP_1_1)) {
+ entity.setChunked(true);
+ }
+ response.setEntity(entity);
+
+ // hand off processing of the request to a thread off the pool
+ workerPool.execute(
+ new ServerWorker(cfgCtx, conn, this,
+ request, Channels.newInputStream(requestPipe.source()),
+ response, Channels.newOutputStream(responsePipe.sink())));
+
+ } catch (IOException e) {
+ handleException("Error processing request received for : " +
+ request.getRequestLine().getUri(), e, conn);
+ }
+ }
+
+ /**
+ * Process ready input by writing it into the Pipe
+ * @param conn the connection being processed
+ * @param decoder the content decoder in use
+ */
+ public void inputReady(final NHttpServerConnection conn, final ContentDecoder decoder) {
+
+ HttpContext context = conn.getContext();
+ Pipe.SinkChannel sink = (Pipe.SinkChannel) context.getAttribute(REQUEST_SINK_CHANNEL);
+ ByteBuffer inbuf = (ByteBuffer) context.getAttribute(REQUEST_BUFFER);
+
+ try {
+ while (decoder.read(inbuf) > 0) {
+ inbuf.flip();
+ sink.write(inbuf);
+ inbuf.compact();
+ }
+
+ if (decoder.isCompleted()) {
+ sink.close();
+ }
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ }
+ }
+
+ /**
+ * Process ready output by writing into the channel
+ * @param conn the connection being processed
+ * @param encoder the content encoder in use
+ */
+ public void outputReady(final NHttpServerConnection conn, final ContentEncoder encoder) {
+
+ HttpContext context = conn.getContext();
+ HttpResponse response = conn.getHttpResponse();
+ Pipe.SourceChannel source = (Pipe.SourceChannel) context.getAttribute(RESPONSE_SOURCE_CHANNEL);
+ ByteBuffer outbuf = (ByteBuffer) context.getAttribute(RESPONSE_BUFFER);
+
+ try {
+ int bytesRead = source.read(outbuf);
+ if (bytesRead == -1) {
+ encoder.complete();
+ } else {
+ outbuf.flip();
+ encoder.write(outbuf);
+ outbuf.compact();
+ }
+
+ if (encoder.isCompleted()) {
+ source.close();
+ if (!connStrategy.keepAlive(response, context)) {
+ conn.close();
+ }
+ }
+
+ } catch (IOException e) {
+ handleException("I/O Error : " + e.getMessage(), e, conn);
+ }
+ }
+
+ /**
+ * Commit the response to the connection. Processes the response through the configured
+ * HttpProcessor and submits it to be sent out
+ * @param conn the connection being processed
+ * @param response the response to commit over the connection
+ */
+ public void commitResponse(final NHttpServerConnection conn, final HttpResponse response) {
+ try {
+ httpProcessor.process(response, conn.getContext());
+ conn.submitResponse(response);
+ } catch (HttpException e) {
+ handleException("Unexpected HTTP protocol error : " + e.getMessage(), e, conn);
+ } catch (IOException e) {
+ handleException("IO error submiting response : " + e.getMessage(), e, conn);
+ }
+ }
+
+
+ /**
+ * Handle connection timeouts by shutting down the connections
+ * @param conn the connection being processed
+ */
+ public void timeout(final NHttpServerConnection conn) {
+ HttpRequest req = (HttpRequest) conn.getContext().getAttribute(HttpContext.HTTP_REQUEST);
+ log.warn("Connection Timeout for request to : " + req.getRequestLine().getUri());
+ shutdownConnection(conn);
+ }
+
+ public void connected(final NHttpServerConnection conn) {
+ log.trace("New incoming connection");
+ }
+
+ public void closed(final NHttpServerConnection conn) {
+ log.trace("Connection closed");
+ }
+
+ /**
+ * Handle HTTP Protocol violations with an error response
+ * @param conn the connection being processed
+ * @param e the exception encountered
+ */
+ public void exception(final NHttpServerConnection conn, final HttpException e) {
+ HttpRequest request = conn.getHttpRequest();
+ HttpVersion ver = request.getRequestLine().getHttpVersion();
+ HttpResponse response = responseFactory.newHttpResponse(ver, HttpStatus.SC_BAD_REQUEST);
+ byte[] msg = EncodingUtils.getAsciiBytes("Malformed HTTP request: " + e.getMessage());
+ ByteArrayEntity entity = new ByteArrayEntity(msg);
+ entity.setContentType("text/plain; charset=US-ASCII");
+ response.setEntity(entity);
+ commitResponse(conn, response);
+ }
+
+ /**
+ * Handle IO errors while reading or writing to underlying channels
+ * @param conn the connection being processed
+ * @param e the exception encountered
+ */
+ public void exception(NHttpServerConnection conn, IOException e) {
+ log.error("I/O error: " + e.getMessage());
+ shutdownConnection(conn);
+ }
+
+ // ----------- utility methods -----------
+
+ private void handleException(String msg, Exception e, NHttpServerConnection conn) {
+ log.error(msg, e);
+ if (conn != null) {
+ shutdownConnection(conn);
+ }
+ }
+
+ /**
+ * Shutdown the connection ignoring any IO errors during the process
+ * @param conn the connection to be shutdown
+ */
+ private void shutdownConnection(final HttpConnection conn) {
+ try {
+ conn.shutdown();
+ } catch (IOException ignore) {}
+ }
+
+ /**
+ * Return the HttpProcessor for responses
+ * @return the HttpProcessor that processes HttpResponses of this server
+ */
+ private HttpProcessor getHttpProcessor() {
+ BasicHttpProcessor httpProcessor = new BasicHttpProcessor();
+ httpProcessor.addInterceptor(new ResponseDate());
+ httpProcessor.addInterceptor(new ResponseServer());
+ httpProcessor.addInterceptor(new ResponseContent());
+ httpProcessor.addInterceptor(new ResponseConnControl());
+ return httpProcessor;
+ }
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerWorker.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerWorker.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerWorker.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/ServerWorker.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.Constants;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.transport.http.HTTPTransportUtils;
+import org.apache.axis2.util.UUIDGenerator;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.Header;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.nio.NHttpServerConnection;
+import org.apache.http.protocol.HTTP;
+
+import javax.xml.namespace.QName;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Processes an incoming request through Axis2. An instance of this class would be created to
+ * process each unique request
+ */
+public class ServerWorker implements Runnable {
+
+ private static final Log log = LogFactory.getLog(ServerWorker.class);
+
+ /** the incoming message to be processed */
+ private MessageContext msgContext = null;
+ /** the Axis2 configuration context */
+ private ConfigurationContext cfgCtx = null;
+ /** the message handler to be used */
+ private ServerHandler serverHandler = null;
+ /** the underlying http connection */
+ private NHttpServerConnection conn = null;
+ /** the http request */
+ private HttpRequest request = null;
+ /** the http response message (which the this would be creating) */
+ private HttpResponse response = null;
+ /** the input stream to read the incoming message body */
+ private InputStream is = null;
+ /** the output stream to write the response message body */
+ private OutputStream os = null;
+ private static final String SOAPACTION = "SOAPAction";
+
+ /**
+ * Create a new server side worker to process an incoming message and optionally begin creating
+ * its output. This however does not force the processor to write a response back as the
+ * traditional servlet service() method, but creates the background required to write the
+ * response, if one would be created.
+ * @param cfgCtx the Axis2 configuration context
+ * @param conn the underlying http connection
+ * @param serverHandler the handler of the server side messages
+ * @param request the http request received (might still be in the process of being streamed)
+ * @param is the stream input stream to read the request body
+ * @param response the response to be populated if applicable
+ * @param os the output stream to write the response body if one is applicable
+ */
+ public ServerWorker(final ConfigurationContext cfgCtx, final NHttpServerConnection conn,
+ final ServerHandler serverHandler,
+ final HttpRequest request, final InputStream is,
+ final HttpResponse response, final OutputStream os) {
+
+ this.cfgCtx = cfgCtx;
+ this.conn = conn;
+ this.serverHandler = serverHandler;
+ this.request = request;
+ this.response = response;
+ this.is = is;
+ this.os = os;
+ this.msgContext = createMessageContext(request);
+ }
+
+ /**
+ * Create an Axis2 message context for the given http request. The request may be in the
+ * process of being streamed
+ * @param request the http request to be used to create the corresponding Axis2 message context
+ * @return the Axis2 message context created
+ */
+ private MessageContext createMessageContext(HttpRequest request) {
+
+ MessageContext msgContext = new MessageContext();
+ msgContext.setProperty(MessageContext.TRANSPORT_NON_BLOCKING, Boolean.TRUE);
+ msgContext.setConfigurationContext(cfgCtx);
+ msgContext.setIncomingTransportName(Constants.TRANSPORT_HTTP);
+ msgContext.setProperty(Constants.OUT_TRANSPORT_INFO, this);
+ msgContext.setServiceGroupContextId(UUIDGenerator.getUUID());
+ msgContext.setServerSide(true);
+ msgContext.setProperty(
+ Constants.Configuration.TRANSPORT_IN_URL, request.getRequestLine().getUri());
+
+ Map headers = new HashMap();
+ Header[] headerArr = request.getAllHeaders();
+ for (int i = 0; i < headerArr.length; i++) {
+ headers.put(headerArr[i].getName(), headerArr[i].getValue());
+ }
+ msgContext.setProperty(MessageContext.TRANSPORT_HEADERS, headers);
+
+ try {
+ msgContext.setTransportOut(cfgCtx.getAxisConfiguration()
+ .getTransportOut(new QName(Constants.TRANSPORT_HTTP)));
+ msgContext.setTransportIn(cfgCtx.getAxisConfiguration()
+ .getTransportIn(new QName(Constants.TRANSPORT_HTTP)));
+ } catch (AxisFault af) {
+ handleException("Unable to get out/in http transport configurations from Axis2", af);
+ return null;
+ }
+
+ return msgContext;
+ }
+
+ /**
+ * Process the incoming request
+ */
+ public void run() {
+
+ String method = request.getRequestLine().getMethod().toUpperCase();
+ if ("GET".equals(method)) {
+ //processGet(response);
+ } else if ("POST".equals(method)) {
+ processPost();
+ } else {
+ handleException("Unsupported method : " + method, null);
+ }
+ }
+
+ /**
+ *
+ */
+ private void processPost() {
+
+ try {
+ HTTPTransportUtils.processHTTPPostRequest(
+ msgContext, is,
+ os,
+ (request.getFirstHeader(HTTP.CONTENT_TYPE) != null ?
+ request.getFirstHeader(HTTP.CONTENT_TYPE).getValue() : null),
+ (request.getFirstHeader(SOAPACTION) != null ?
+ request.getFirstHeader(SOAPACTION).getValue() : null),
+ request.getRequestLine().getUri());
+ } catch (AxisFault e) {
+ handleException("Error processing POST request ", e);
+ }
+ }
+
+ private void handleException(String msg, Exception e) {
+
+ log.error(msg, e);
+ if (conn != null) {
+ try {
+ conn.shutdown();
+ } catch (IOException ignore) {}
+ }
+ /*try {
+ AxisEngine engine = new AxisEngine(cfgCtx);
+ msgContext.setProperty(MessageContext.TRANSPORT_OUT, response.getOutputStream());
+ msgContext.setProperty(org.apache.axis2.Constants.OUT_TRANSPORT_INFO, response.getOutputStream());
+ MessageContext faultContext = engine.createFaultMessageContext(msgContext, e);
+ engine.sendFault(faultContext);
+
+ } catch (Exception ex) {
+ response.addHeader(Constants.CONTENT_TYPE, Constants.TEXT_PLAIN);
+ OutputStreamWriter out = new OutputStreamWriter(
+ response.getOutputStream());
+ try {
+ out.write(ex.getMessage());
+ out.close();
+ } catch (IOException ee) {
+ }
+
+ } finally {
+ response.setStatus(ResponseStatus.INTERNAL_SERVER_ERROR);
+ response.commit();
+ }*/
+ }
+
+
+ public HttpResponse getResponse() {
+ return response;
+ }
+
+ public OutputStream getOutputStream() {
+ return os;
+ }
+
+ public ServerHandler getServiceHandler() {
+ return serverHandler;
+ }
+
+ public NHttpServerConnection getConn() {
+ return conn;
+ }
+}
Added: webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Util.java
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Util.java?view=auto&rev=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Util.java (added)
+++ webservices/synapse/trunk/java/modules/nhttp/src/org/apache/axis2/transport/nhttp/Util.java Mon Jan 22 02:34:36 2007
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.axis2.transport.nhttp;
+
+import org.apache.axis2.addressing.EndpointReference;
+import org.apache.axis2.addressing.AddressingConstants;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.context.OperationContext;
+import org.apache.axis2.Constants;
+import org.apache.axis2.transport.http.HTTPTransportUtils;
+import org.apache.axis2.transport.http.HTTPConstants;
+import org.apache.axiom.om.OMOutputFormat;
+
+public class Util {
+
+ /**
+ * Get the EPR for the message passed in
+ * @param msgContext the message context
+ * @return the destination EPR
+ */
+ public static EndpointReference getDestinationEPR(MessageContext msgContext) {
+
+ // Trasnport URL can be different from the WSA-To
+ String transportURL = (String) msgContext.getProperty(
+ Constants.Configuration.TRANSPORT_URL);
+
+ if (transportURL != null) {
+ return new EndpointReference(transportURL);
+ } else if (
+ (msgContext.getTo() != null) &&
+ !AddressingConstants.Submission.WSA_ANONYMOUS_URL.equals(
+ msgContext.getTo().getAddress()) &&
+ !AddressingConstants.Final.WSA_ANONYMOUS_URL.equals(
+ msgContext.getTo().getAddress())) {
+ return msgContext.getTo();
+ }
+ return null;
+ }
+
+ /**
+ * Retirn the OMOutputFormat to be used for the message context passed in
+ * @param msgContext the message context
+ * @return the OMOutputFormat to be used
+ */
+ public static OMOutputFormat getOMOutputFormat(MessageContext msgContext) {
+
+ OMOutputFormat format = new OMOutputFormat();
+ String charSetEnc = (String) msgContext.getProperty(
+ Constants.Configuration.CHARACTER_SET_ENCODING);
+
+ if (charSetEnc == null) {
+ OperationContext opctx = msgContext.getOperationContext();
+ if (opctx != null) {
+ charSetEnc = (String) opctx.getProperty(
+ Constants.Configuration.CHARACTER_SET_ENCODING);
+ }
+ }
+
+ // if the charset encoding is still not found use the default
+ if (charSetEnc == null) {
+ charSetEnc = MessageContext.DEFAULT_CHAR_SET_ENCODING;
+ }
+ format.setCharSetEncoding(charSetEnc);
+
+ msgContext.setDoingMTOM(HTTPTransportUtils.doWriteMTOM(msgContext));
+ msgContext.setDoingSwA(HTTPTransportUtils.doWriteSwA(msgContext));
+ msgContext.setDoingREST(HTTPTransportUtils.isDoingREST(msgContext));
+ format.setSOAP11(msgContext.isSOAP11());
+ format.setDoOptimize(msgContext.isDoingMTOM());
+ return format;
+ }
+
+ /**
+ * Get the content type for the message passed in
+ * @param msgContext the message
+ * @return content type of the message
+ */
+ public static String getContentType(MessageContext msgContext) {
+ Object contentTypeObject = msgContext.getProperty(Constants.Configuration.CONTENT_TYPE);
+ if (contentTypeObject != null) {
+ return (String) contentTypeObject;
+ } else if (msgContext.isDoingREST()) {
+ return HTTPConstants.MEDIA_TYPE_APPLICATION_XML;
+ } else {
+ return getOMOutputFormat(msgContext).getContentType();
+ }
+ }
+
+}
Modified: webservices/synapse/trunk/java/modules/samples/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/modules/samples/pom.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/modules/samples/pom.xml (original)
+++ webservices/synapse/trunk/java/modules/samples/pom.xml Mon Jan 22 02:34:36 2007
@@ -26,7 +26,7 @@
<parent>
<groupId>org.apache.synapse</groupId>
<artifactId>Apache-Synapse</artifactId>
- <version>0.91</version>
+ <version>SNAPSHOT</version>
</parent>
<groupId>org.apache.synapse</groupId>
Modified: webservices/synapse/trunk/java/pom.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/pom.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/pom.xml (original)
+++ webservices/synapse/trunk/java/pom.xml Mon Jan 22 02:34:36 2007
@@ -26,7 +26,7 @@
<groupId>org.apache.synapse</groupId>
<artifactId>Apache-Synapse</artifactId>
- <version>0.91</version>
+ <version>SNAPSHOT</version>
<name>Apache Synapse</name>
<description>Apache Synapse</description>
@@ -219,6 +219,11 @@
<artifactId>synapse-samples</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.synapse</groupId>
+ <artifactId>synapse-nhttp</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
@@ -404,10 +409,22 @@
<version>${backport_util_concurrent.version}</version>
</dependency>
- <dependency>
+ <!--<dependency>
<groupId>httpcomponents-httpcore</groupId>
<artifactId>jakarta-httpcore</artifactId>
<version>${jakarta.httpcore.version}</version>
+ </dependency>-->
+
+ <!-- dependencies for nhttp transport -->
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>jakarta-httpcore</artifactId>
+ <version>${jakarta.httpcore.nio.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>jakarta-httpcore-nio</artifactId>
+ <version>${jakarta.httpcore.nio.version}</version>
</dependency>
</dependencies>
@@ -559,7 +576,7 @@
</repositories>
<modules>
- <!--<module>modules/nhttp</module>-->
+ <module>modules/nhttp</module>
<module>modules/core</module>
<module>modules/extensions</module>
<module>modules/samples</module>
@@ -567,7 +584,7 @@
</modules>
<properties>
- <synapse.version>0.91</synapse.version>
+ <synapse.version>SNAPSHOT</synapse.version>
<axis2.version>1.1.051</axis2.version>
<rampart.version>1.1</rampart.version>
<axiom.version>1.2.1</axiom.version>
@@ -584,7 +601,8 @@
<commons.httpclient.version>3.0.1</commons.httpclient.version>
<commons.codec.version>1.3</commons.codec.version>
<commons.fileupload.version>1.1.1</commons.fileupload.version>
- <jakarta.httpcore.version>4.0-alpha2</jakarta.httpcore.version>
+ <!--<jakarta.httpcore.version>4.0-alpha2</jakarta.httpcore.version>-->
+ <jakarta.httpcore.nio.version>4.0-alpha4-SNAPSHOT</jakarta.httpcore.nio.version>
<log4j.version>1.2.13</log4j.version>
<junit.version>3.8.2</junit.version>
Modified: webservices/synapse/trunk/java/repository/conf/axis2.xml
URL: http://svn.apache.org/viewvc/webservices/synapse/trunk/java/repository/conf/axis2.xml?view=diff&rev=498575&r1=498574&r2=498575
==============================================================================
--- webservices/synapse/trunk/java/repository/conf/axis2.xml (original)
+++ webservices/synapse/trunk/java/repository/conf/axis2.xml Mon Jan 22 02:34:36 2007
@@ -103,9 +103,9 @@
<!-- Transport Ins -->
<!-- ================================================= -->
<!--<transportReceiver name="http" class="org.apache.axis2.transport.nhttp.AsyncHTTPListener"/>-->
- <transportReceiver name="http"
+ <!--<transportReceiver name="http"
class="org.apache.axis2.transport.http.SimpleHTTPServer">
- <parameter name="port" locked="false">8080</parameter>
+ <parameter name="port" locked="false">8080</parameter>-->
<!-- Here is the complete list of supported parameters (see example settings further below):
port: the port to listen on (default 6060)
hostname: if non-null, url prefix used in reply-to endpoint references (default null)
@@ -128,6 +128,12 @@
<!-- <parameter name="RequestMaxThreadPoolSize" locked="false">100</parameter> -->
<!-- <parameter name="threadKeepAliveTime" locked="false">240000</parameter> -->
<!-- <parameter name="threadKeepAliveTimeUnit" locked="false">MILLISECONDS</parameter> -->
+ <!--</transportReceiver>-->
+
+ <!-- the experimental non blocking http transport based on HttpCore + NIO extensions -->
+ <transportReceiver name="http" class="org.apache.axis2.transport.nhttp.HttpCoreNIOListener">
+ <parameter name="port" locked="false">8080</parameter>
+ <parameter name="non-blocking" locked="false">true</parameter>
</transportReceiver>
<!--Uncomment this and configure as appropriate for JMS transport support, after setting up your JMS environment (e.g. ActiveMQ)
@@ -176,13 +182,13 @@
class="org.apache.axis2.transport.tcp.TCPTransportSender"/>
<transportSender name="local"
class="org.apache.axis2.transport.local.LocalTransportSender"/>
- <transportSender name="http"
- class="org.apache.axis2.transport.http.CommonsHTTPTransportSender"/>
<!--<transportSender name="http"
- class="org.apache.axis2.transport.nhttp.AsyncHTTPSender">
- <parameter name="PROTOCOL" locked="false">HTTP/1.1</parameter>
- <parameter name="Transfer-Encoding" locked="false">chunked</parameter>
- </transportSender>-->
+ class="org.apache.axis2.transport.http.CommonsHTTPTransportSender"/>-->
+ <!-- the experimental non-blocking http transport based on HttpCore + NIO extensions -->
+ <transportSender name="http"
+ class="org.apache.axis2.transport.nhttp.HttpCoreNIOSender">
+ <parameter name="non-blocking" locked="false">true</parameter>
+ </transportSender>
<transportSender name="https"
class="org.apache.axis2.transport.http.CommonsHTTPTransportSender">
<parameter name="PROTOCOL" locked="false">HTTP/1.1</parameter>
---------------------------------------------------------------------
To unsubscribe, e-mail: synapse-dev-unsubscribe@ws.apache.org
For additional commands, e-mail: synapse-dev-help@ws.apache.org