You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@synapse.apache.org by hi...@apache.org on 2013/08/14 23:35:25 UTC
svn commit: r1514059 - in /synapse/trunk/java/modules:
core/src/test/java/org/apache/synapse/registry/
transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/
transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/...
Author: hiranya
Date: Wed Aug 14 21:35:25 2013
New Revision: 1514059
URL: http://svn.apache.org/r1514059
Log:
Handling the endOfInput property in the PT SourceHandler - Otherwise JMX stats get messed up.
Modified:
synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/registry/DynamicResourceTest.java
synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceContext.java
synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java
synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java
Modified: synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/registry/DynamicResourceTest.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/registry/DynamicResourceTest.java?rev=1514059&r1=1514058&r2=1514059&view=diff
==============================================================================
--- synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/registry/DynamicResourceTest.java (original)
+++ synapse/trunk/java/modules/core/src/test/java/org/apache/synapse/registry/DynamicResourceTest.java Wed Aug 14 21:35:25 2013
@@ -101,7 +101,7 @@ public class DynamicResourceTest extends
System.out.println("Testing advanced sequence caching...");
synCtx = TestUtils.createLightweightSynapseMessageContext("<empty/>", config);
System.out.println("Waiting for the cache to expire...");
- Thread.sleep(8500L);
+ Thread.sleep(10000L);
Mediator seq3 = synCtx.getSequence(KEY_DYNAMIC_SEQUENCE_1);
assertNotNull(seq3);
assertTrue(((SequenceMediator) seq3).isInitialized());
@@ -114,7 +114,7 @@ public class DynamicResourceTest extends
System.out.println("Testing sequence reloading...");
registry.updateResource(KEY_DYNAMIC_SEQUENCE_1, TestUtils.createOMElement(DYNAMIC_SEQUENCE_2));
System.out.println("Waiting for the cache to expire...");
- Thread.sleep(8500L);
+ Thread.sleep(10000L);
synCtx = TestUtils.createLightweightSynapseMessageContext("<empty/>", config);
Mediator seq4 = synCtx.getSequence(KEY_DYNAMIC_SEQUENCE_1);
assertNotNull(seq4);
@@ -158,7 +158,7 @@ public class DynamicResourceTest extends
System.out.println("Testing advanced endpoint caching...");
synCtx = TestUtils.createSynapseMessageContext("<empty/>", config);
System.out.println("Waiting for the cache to expire...");
- Thread.sleep(8500L);
+ Thread.sleep(10000L);
Endpoint ep3 = synCtx.getEndpoint(KEY_DYNAMIC_ENDPOINT_1);
assertNotNull(ep3);
assertEquals(1, registry.getHitCount());
@@ -168,7 +168,7 @@ public class DynamicResourceTest extends
System.out.println("Testing endpoint reloading...");
registry.updateResource(KEY_DYNAMIC_ENDPOINT_1, TestUtils.createOMElement(DYNAMIC_ENDPOINT_2));
System.out.println("Waiting for the cache to expire...");
- Thread.sleep(8500L);
+ Thread.sleep(10000L);
synCtx = TestUtils.createSynapseMessageContext("<empty/>", config);
Endpoint ep4 = synCtx.getEndpoint(KEY_DYNAMIC_ENDPOINT_1);
assertNotNull(ep4);
Modified: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceContext.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceContext.java?rev=1514059&r1=1514058&r2=1514059&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceContext.java (original)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceContext.java Wed Aug 14 21:35:25 2013
@@ -78,7 +78,11 @@ public class SourceContext {
public void reset() {
this.request = null;
this.response = null;
- this.state = ProtocolState.REQUEST_READY;
+ if (this.state != ProtocolState.CLOSED) {
+ // if the connection is not closed yet, prepare to receive a new request
+ // on it
+ this.state = ProtocolState.REQUEST_READY;
+ }
if (writer != null) {
ByteBuffer buffer = writer.getBuffer();
Modified: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java?rev=1514059&r1=1514058&r2=1514059&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java (original)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/SourceHandler.java Wed Aug 14 21:35:25 2013
@@ -250,16 +250,31 @@ public class SourceHandler implements NH
metrics.incrementBytesSent(bytesSent);
} catch (IOException e) {
logIOException(e);
-
informWriterError(conn);
-
SourceContext.updateState(conn, ProtocolState.CLOSING);
sourceConfiguration.getSourceConnections().shutDownConnection(conn);
}
}
public void endOfInput(NHttpServerConnection conn) throws IOException {
- closed(conn);
+ ProtocolState state = SourceContext.getState(conn);
+
+ if (state == ProtocolState.REQUEST_READY || state == ProtocolState.RESPONSE_DONE) {
+ if (log.isDebugEnabled()) {
+ log.debug("Keep-Alive connection was closed at the remote end: " + conn);
+ }
+ } else if (state == ProtocolState.REQUEST_BODY || state == ProtocolState.REQUEST_HEAD) {
+ informReaderError(conn);
+ log.warn("Connection closed at the remote end while reading the request: " + conn);
+ } else if (state == ProtocolState.RESPONSE_BODY || state == ProtocolState.RESPONSE_HEAD) {
+ informWriterError(conn);
+ log.warn("Connection closed at the remote end while writing the response: " + conn);
+ } else if (state == ProtocolState.REQUEST_DONE) {
+ log.warn("Connection closed by the client after request is read: " + conn);
+ }
+
+ SourceContext.updateState(conn, ProtocolState.CLOSED);
+ sourceConfiguration.getSourceConnections().shutDownConnection(conn);
}
public void exception(NHttpServerConnection conn, Exception e) {
@@ -361,10 +376,9 @@ public class SourceHandler implements NH
try {
sourceConfiguration.getHttpProcessor().process(response, httpContext);
conn.submitResponse(response);
- SourceContext.updateState(conn, ProtocolState.CLOSED);
- conn.close();
} catch (Exception ex) {
log.error("Error while handling HttpException", ex);
+ } finally {
SourceContext.updateState(conn, ProtocolState.CLOSED);
sourceConfiguration.getSourceConnections().shutDownConnection(conn);
}
@@ -403,12 +417,10 @@ public class SourceHandler implements NH
if (log.isDebugEnabled()) {
log.debug("Keep-Alive connection was closed: " + conn);
}
- } else if (state == ProtocolState.REQUEST_BODY ||
- state == ProtocolState.REQUEST_HEAD) {
+ } else if (state == ProtocolState.REQUEST_BODY || state == ProtocolState.REQUEST_HEAD) {
informReaderError(conn);
log.warn("Connection closed while reading the request: " + conn);
- } else if (state == ProtocolState.RESPONSE_BODY ||
- state == ProtocolState.RESPONSE_HEAD) {
+ } else if (state == ProtocolState.RESPONSE_BODY || state == ProtocolState.RESPONSE_HEAD) {
informWriterError(conn);
log.warn("Connection closed while writing the response: " + conn);
} else if (state == ProtocolState.REQUEST_DONE) {
@@ -416,9 +428,10 @@ public class SourceHandler implements NH
}
metrics.disconnected();
-
- SourceContext.updateState(conn, ProtocolState.CLOSED);
- sourceConfiguration.getSourceConnections().shutDownConnection(conn);
+ if (state != ProtocolState.CLOSED) {
+ SourceContext.updateState(conn, ProtocolState.CLOSED);
+ sourceConfiguration.getSourceConnections().shutDownConnection(conn);
+ }
}
private void handleInvalidState(NHttpServerConnection conn, String action) {
Modified: synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java
URL: http://svn.apache.org/viewvc/synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java?rev=1514059&r1=1514058&r2=1514059&view=diff
==============================================================================
--- synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java (original)
+++ synapse/trunk/java/modules/transports/core/nhttp/src/main/java/org/apache/synapse/transport/passthru/jmx/LatencyView.java Wed Aug 14 21:35:25 2013
@@ -46,7 +46,7 @@ import java.util.concurrent.atomic.Atomi
*
* <ul>
* <li>t1 - Receiving a new request (ServerHandler#requestReceived)</li>
- * <li>t2 - Obtaining a connection to forward the request (Clienthandler#processConnection)</li>
+ * <li>t2 - Obtaining a connection to forward the request (ClientHandler#processConnection)</li>
* <li>t3 - Reading the complete response from the backend server (ClientHandler#inputReady)</li>
* <li>t4 - Writing the complete response to the client (ServerHandler#outputReady)</li>
* <ul>
@@ -58,7 +58,7 @@ import java.util.concurrent.atomic.Atomi
*/
public class LatencyView implements LatencyViewMBean {
- private static final String NHTTP_LATENCY_VIEW = "NhttpTransportLatency";
+ private static final String NHTTP_LATENCY_VIEW = "PassThroughTransportLatency";
private static final int SMALL_DATA_COLLECTION_PERIOD = 5;
private static final int LARGE_DATA_COLLECTION_PERIOD = 5 * 60;
@@ -124,7 +124,7 @@ public class LatencyView implements Late
*
* @param reqArrival The request arrival time
* @param reqDeparture The request departure time (backend connection establishment)
- * @param resArrival The resoponse arrival time
+ * @param resArrival The response arrival time
* @param resDeparture The response departure time
*/
public void notifyTimes(long reqArrival, long reqDeparture,