merging the improvement https://issues.apache.org/jira/browse/SYNAPSE-1006
git-svn-id: https://svn.apache.org/repos/asf/synapse/trunk@1745921 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ClientHandler.java b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ClientHandler.java
index 9eadfe2..3585e17 100644
--- a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ClientHandler.java
+++ b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ClientHandler.java
@@ -279,14 +279,15 @@
if (axis2Request != null && !axis2Request.isCompleted()) {
checkAxisRequestComplete(conn, NhttpConstants.CONNECTION_CLOSED, message, null);
+ shutdownConnection(conn, true, "Connection closed before response is received");
} else {
if (log.isDebugEnabled()) {
log.debug(getErrorMessage("Keep-alive connection closed", conn));
}
+ shutdownConnection(conn, false, null);
}
HttpContext context = conn.getContext();
- shutdownConnection(conn);
context.removeAttribute(RESPONSE_SINK_BUFFER);
context.removeAttribute(REQUEST_SOURCE_BUFFER);
metrics.disconnected();
@@ -309,14 +310,15 @@
if (axis2Request != null && !axis2Request.isCompleted()) {
checkAxisRequestComplete(conn, NhttpConstants.CONNECTION_TIMEOUT, message, null);
+ shutdownConnection(conn, true, "Connection timeout before response is received");
} else {
if (log.isDebugEnabled()) {
log.debug(getErrorMessage("Keep-alive connection timed out", conn));
}
+ shutdownConnection(conn, false, null);
}
HttpContext context = conn.getContext();
- shutdownConnection(conn);
context.removeAttribute(RESPONSE_SINK_BUFFER);
context.removeAttribute(REQUEST_SOURCE_BUFFER);
}
@@ -332,7 +334,7 @@
exception(conn, (IOException) e);
} else {
log.error(e.getMessage(), e);
- shutdownConnection(conn);
+ shutdownConnection(conn, true, "Error occurred : " + e.getMessage());
}
}
@@ -346,7 +348,7 @@
String message = getErrorMessage("HTTP protocol violation : " + e.getMessage(), conn);
log.error(message, e);
checkAxisRequestComplete(conn, NhttpConstants.PROTOCOL_VIOLATION, message, e);
- shutdownConnection(conn);
+ shutdownConnection(conn, true, "HTTP protocol violation : " + e.getMessage());
}
/**
@@ -363,7 +365,7 @@
log.error(message, e);
}
checkAxisRequestComplete(conn, NhttpConstants.SND_IO_ERROR_SENDING, message, e);
- shutdownConnection(conn);
+ shutdownConnection(conn, true, "I/O error : " + e.getMessage());
}
/**
@@ -547,12 +549,12 @@
try {
// this is a connection we should not re-use
ConnectionPool.forget(conn);
- shutdownConnection(conn);
+ shutdownConnection(conn, false, null);
context.removeAttribute(RESPONSE_SINK_BUFFER);
context.removeAttribute(REQUEST_SOURCE_BUFFER);
} catch (Exception ignore) {}
} else if (!connStrategy.keepAlive(response, context)) {
- shutdownConnection(conn);
+ shutdownConnection(conn, false, null);
context.removeAttribute(RESPONSE_SINK_BUFFER);
context.removeAttribute(REQUEST_SOURCE_BUFFER);
} else {
@@ -957,7 +959,7 @@
try {
// this is a connection we should not re-use
ConnectionPool.forget(conn);
- shutdownConnection(conn);
+ shutdownConnection(conn, false, null);
context.removeAttribute(RESPONSE_SINK_BUFFER);
context.removeAttribute(REQUEST_SOURCE_BUFFER);
} catch (Exception ignore) {}
@@ -976,19 +978,28 @@
workerPool.execute(task);
}
+
/**
* Shutdown the connection ignoring any IO errors during the process
- *
+ *
* @param conn the connection to be shutdown
+ * @param isError whether shutdown is due to an error
+ * @param errorMsg error message if shutdown happens on error
*/
- private void shutdownConnection(final NHttpClientConnection conn) {
+ private void shutdownConnection(final NHttpClientConnection conn, boolean isError, String errorMsg) {
if (conn instanceof HttpInetConnection) {
HttpInetConnection inetConnection = (HttpInetConnection) conn;
- if (log.isDebugEnabled()) {
- log.debug("Connection to remote address : " + inetConnection.getRemoteAddress()
+ if (log.isWarnEnabled() && (isError || log.isDebugEnabled())) {
+ String msg = "Connection to remote address : " + inetConnection.getRemoteAddress()
+ ":" + inetConnection.getRemotePort() + " from local address : "
+ inetConnection.getLocalAddress() + ":" + inetConnection.getLocalPort() +
- " is closed!");
+ " is closed!"
+ + (errorMsg != null ? " - On error : " + errorMsg : "");
+ if (isError) {
+ log.warn(msg);
+ } else {
+ log.debug(msg);
+ }
}
if (countConnections) {
@@ -1131,7 +1142,7 @@
log.error(msg, e);
}
if (conn != null) {
- shutdownConnection(conn);
+ shutdownConnection(conn, true, msg);
}
}
diff --git a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/NhttpConstants.java b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/NhttpConstants.java
index 5e17947..1967cd4 100644
--- a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/NhttpConstants.java
+++ b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/NhttpConstants.java
@@ -53,6 +53,9 @@
public static final String REQUEST_CONTENT_TYPE = "DEFAULT_REQUEST_CONTENT_TYPE";
public static final String DEFAULT_CONTENT_TYPE = "application/octet-stream";
+ public static final String CLIENT_REMOTE_ADDR = "CLIENT_REMOTE_ADDR";
+ public static final String CLIENT_REMOTE_PORT = "CLIENT_REMOTE_PORT";
+
public static final String SEND_TIMEOUT = "SEND_TIMEOUT";
public static final String HIDDEN_SERVICE_PARAM_NAME = "hiddenService";
diff --git a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ServerHandler.java b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ServerHandler.java
index 164afff..8305e0d 100644
--- a/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ServerHandler.java
+++ b/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/nhttp/ServerHandler.java
@@ -53,6 +53,7 @@
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.net.InetAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -275,7 +276,7 @@
if (outBuf == null) {
// fix for SYNAPSE 584. This is a temporarily fix because of HTTPCORE-208
- shutdownConnection(conn);
+ shutdownConnection(conn, false, null);
return;
}
@@ -379,10 +380,10 @@
conn.submitResponse(response);
} catch (HttpException e) {
- shutdownConnection(conn);
+ shutdownConnection(conn, true, e.getMessage());
throw e;
} catch (IOException e) {
- shutdownConnection(conn);
+ shutdownConnection(conn, true, e.getMessage());
throw e;
}
}
@@ -399,13 +400,15 @@
if (log.isDebugEnabled()) {
log.debug("Keepalive connection was closed");
}
+ shutdownConnection(conn, false, null);
} else {
- log.error("Connection Timeout - before message body was fully read : " + conn);
+ String msg = "Connection Timeout - before message body was fully read : " + conn;
+ log.error(msg);
if (metrics != null) {
metrics.incrementTimeoutsReceiving();
}
+ shutdownConnection(conn, true, msg);
}
- shutdownConnection(conn);
}
public void connected(final NHttpServerConnection conn) {
@@ -453,7 +456,7 @@
public void closed(final NHttpServerConnection conn) {
HttpContext context = conn.getContext();
- shutdownConnection(conn);
+ shutdownConnection(conn, false, null);
context.removeAttribute(REQUEST_SINK_BUFFER);
context.removeAttribute(RESPONSE_SOURCE_BUFFER);
context.removeAttribute(CONNECTION_CREATION_TIME);
@@ -517,11 +520,12 @@
} else if (e instanceof IOException) {
exception(conn, (IOException) e);
} else {
- log.error("Unexpected error: " + e.getClass().getName(), e);
+ String errMsg = "Unexpected error: " + e.getClass().getName();
+ log.error(errMsg, e);
if (metrics != null) {
metrics.incrementFaultsReceiving();
}
- shutdownConnection(conn);
+ shutdownConnection(conn, true, errMsg);
}
}
@@ -531,31 +535,36 @@
* @param e the exception encountered
*/
public void exception(NHttpServerConnection conn, IOException e) {
+ String errMsg = "I/O error : " + e.getMessage();
if (e instanceof ConnectionClosedException || (e.getMessage() != null &&
e.getMessage().contains("Connection reset by peer") ||
e.getMessage().contains("forcibly closed"))) {
if (log.isDebugEnabled()) {
- log.debug("I/O error (Probably the keepalive connection " +
- "was closed):" + e.getMessage());
+ errMsg = "I/O error (Probably the keepalive connection " +
+ "was closed):" + e.getMessage();
+ log.debug(errMsg);
}
} else if (e.getMessage() != null) {
- String msg = e.getMessage().toLowerCase();
- if (msg.indexOf("broken") != -1) {
- log.warn("I/O error (Probably the connection " +
- "was closed by the remote party):" + e.getMessage());
+ errMsg = e.getMessage().toLowerCase();
+ if (errMsg.indexOf("broken") != -1) {
+ errMsg = "I/O error (Probably the connection " +
+ "was closed by the remote party):" + e.getMessage();
+ log.warn(errMsg);
} else {
- log.error("I/O error: " + e.getMessage(), e);
+ errMsg = "I/O error: " + e.getMessage();
+ log.error(errMsg, e);
}
if (metrics != null) {
metrics.incrementFaultsReceiving();
}
} else {
- log.error("Unexpected I/O error: " + e.getClass().getName(), e);
+ errMsg = "Unexpected I/O error: " + e.getMessage();
+ log.error(errMsg, e);
if (metrics != null) {
metrics.incrementFaultsReceiving();
}
}
- shutdownConnection(conn);
+ shutdownConnection(conn, true, errMsg);
}
// ----------- utility methods -----------
@@ -563,15 +572,17 @@
private void handleException(String msg, Exception e, NHttpServerConnection conn) {
log.error(msg, e);
if (conn != null) {
- shutdownConnection(conn);
+ shutdownConnection(conn, true, e.getMessage());
}
}
/**
* Shutdown the connection ignoring any IO errors during the process
* @param conn the connection to be shutdown
+ * @param isError whether shutdown is due to an error
+ * @param errorMsg error message if shutdown happens on error
*/
- private void shutdownConnection(final NHttpServerConnection conn) {
+ private void shutdownConnection(final NHttpServerConnection conn, boolean isError, String errorMsg) {
SharedOutputBuffer outputBuffer = (SharedOutputBuffer)
conn.getContext().getAttribute(RESPONSE_SOURCE_BUFFER);
if (outputBuffer != null) {
@@ -583,6 +594,39 @@
inputBuffer.close();
}
+ if (log.isWarnEnabled() && (isError || log.isDebugEnabled()) && conn instanceof HttpInetConnection) {
+
+ HttpInetConnection inetConnection = (HttpInetConnection) conn;
+ InetAddress remoteAddress = inetConnection.getRemoteAddress();
+ int remotePort = inetConnection.getRemotePort();
+
+ String msg;
+ if (remotePort != -1 && remoteAddress != null) { // If connection is still alive
+ msg = "Connection from remote address : "
+ + remoteAddress + ":" + remotePort
+ + " to local address : "
+ + inetConnection.getLocalAddress() + ":" + inetConnection.getLocalPort() +
+ " is closed!"
+ + (errorMsg != null ? " - On error : " + errorMsg : "");
+
+ } else { // if connection is already closed. obtain params from http context
+ HttpContext httpContext = conn.getContext();
+ msg = "Connection from remote address : "
+ + httpContext.getAttribute(NhttpConstants.CLIENT_REMOTE_ADDR)
+ + ":" + httpContext.getAttribute(NhttpConstants.CLIENT_REMOTE_PORT)
+ + " to local address : "
+ + inetConnection.getLocalAddress() + ":" + inetConnection.getLocalPort() +
+ " is closed!"
+ + (errorMsg != null ? " - On error : " + errorMsg : "");
+ }
+
+ if (isError) {
+ log.warn(msg);
+ } else {
+ log.debug(msg);
+ }
+ }
+
synchronized (this) {
if (activeConnections.remove(conn) && log.isDebugEnabled()) {
log.debug("Removing the connection : " + conn
diff --git a/java/modules/transports/core/nhttp/src/test/java/org/apache/synapse/transport/nhttp/HttpCoreNIOListenerTest.java b/java/modules/transports/core/nhttp/src/test/java/org/apache/synapse/transport/nhttp/HttpCoreNIOListenerTest.java
index 1a1abbc..cb66341 100644
--- a/java/modules/transports/core/nhttp/src/test/java/org/apache/synapse/transport/nhttp/HttpCoreNIOListenerTest.java
+++ b/java/modules/transports/core/nhttp/src/test/java/org/apache/synapse/transport/nhttp/HttpCoreNIOListenerTest.java
@@ -23,7 +23,6 @@
import junit.framework.TestSuite;
import org.apache.axis2.transport.testkit.ManagedTestSuite;
-import org.apache.axis2.transport.testkit.axis2.TransportDescriptionFactory;
import org.apache.axis2.transport.testkit.http.HttpTransportTestSuiteBuilder;
public class HttpCoreNIOListenerTest extends TestCase {
@@ -33,7 +32,7 @@
// These tests don't work because of a problem similar to SYNAPSE-418
suite.addExclude("(test=EchoXML)");
- TransportDescriptionFactory tdfNIO = new HttpTransportDescriptionFactory();
+ HttpTransportDescriptionFactory tdfNIO = new HttpTransportDescriptionFactory();
HttpTransportTestSuiteBuilder builder = new HttpTransportTestSuiteBuilder(suite, tdfNIO);
