You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tomee.apache.org by rm...@apache.org on 2013/03/26 13:53:31 UTC
svn commit: r1461097 - in /tomee/tomee/trunk/server:
openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java
openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
Author: rmannibucau
Date: Tue Mar 26 12:53:30 2013
New Revision: 1461097
URL: http://svn.apache.org/r1461097
Log:
OPENEJB-2010 forcing openejb http to close the socket after the client read it
Modified:
tomee/tomee/trunk/server/openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java
tomee/tomee/trunk/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
Modified: tomee/tomee/trunk/server/openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java
URL: http://svn.apache.org/viewvc/tomee/tomee/trunk/server/openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java?rev=1461097&r1=1461096&r2=1461097&view=diff
==============================================================================
--- tomee/tomee/trunk/server/openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java (original)
+++ tomee/tomee/trunk/server/openejb-http/src/main/java/org/apache/openejb/server/httpd/OpenEJBHttpServer.java Tue Mar 26 12:53:30 2013
@@ -88,21 +88,11 @@ public class OpenEJBHttpServer implement
@Override
public void service(final Socket socket) throws ServiceException, IOException {
- /**
- * The InputStream used to receive incoming messages from the client.
- */
- InputStream in = null;
- /**
- * The OutputStream used to send outgoing response messages to the client.
- */
- OutputStream out = null;
+ RequestInfos.initRequestInfo(socket);
+ final InputStream in = new CountingInputStream(socket.getInputStream());
+ final OutputStream out = new CountingOutputStream(socket.getOutputStream());
try {
- RequestInfos.initRequestInfo(socket);
-
- in = new CountingInputStream(socket.getInputStream());
- out = new CountingOutputStream(socket.getOutputStream());
-
//TODO: if ssl change to https
final URI socketURI = new URI("http://" + socket.getLocalAddress().getHostAddress() + ":" + socket.getLocalPort());
processRequest(socketURI, in, out);
@@ -110,31 +100,10 @@ public class OpenEJBHttpServer implement
} catch (Throwable e) {
log.error("Unexpected error", e);
} finally {
- if (out != null) {
- try {
- out.flush();
- } catch (Throwable e) {
- //Ignore
- }
- try {
- out.close();
- } catch (Throwable e) {
- //Ignore
- }
- }
-
- if (in != null) {
- try {
- in.close();
- } catch (Throwable e) {
- //Ignore
- }
- }
-
try {
- socket.close();
+ out.flush();
} catch (Throwable e) {
- log.error("Encountered problem while closing connection with client: " + e.getMessage());
+ //Ignore
}
}
}
@@ -160,10 +129,12 @@ public class OpenEJBHttpServer implement
@Override
public void start() throws ServiceException {
+ // no-op
}
@Override
public void stop() throws ServiceException {
+ // no-op
}
@Override
Modified: tomee/tomee/trunk/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java
URL: http://svn.apache.org/viewvc/tomee/tomee/trunk/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java?rev=1461097&r1=1461096&r2=1461097&view=diff
==============================================================================
--- tomee/tomee/trunk/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java (original)
+++ tomee/tomee/trunk/server/openejb-server/src/main/java/org/apache/openejb/server/ServicePool.java Tue Mar 26 12:53:30 2013
@@ -21,12 +21,14 @@ import org.apache.openejb.loader.SystemI
import org.apache.openejb.monitoring.Managed;
import org.apache.openejb.util.LogCategory;
import org.apache.openejb.util.Logger;
+import org.apache.openejb.util.OptionsLog;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
@@ -43,6 +45,9 @@ public class ServicePool extends ServerS
private final ThreadPoolExecutor threadPool;
private final AtomicBoolean stop = new AtomicBoolean();
+ private final CleanUpThread cleanupThread = new CleanUpThread();
+ private long maxResponseSize = Long.MAX_VALUE;
+ private int cleanupTimeout = 500;
public ServicePool(final ServerService next, final Properties properties) {
/**Defaults.
@@ -170,6 +175,37 @@ public class ServicePool extends ServerS
}
@Override
+ public void init(final Properties props) throws Exception {
+ super.init(props);
+ final Options options = new Options(props);
+ options.setLogger(new OptionsLog(log));
+ cleanupTimeout = options.get("cleanupTimeout", 500);
+ maxResponseSize = options.get("maxResponseSize", Long.MAX_VALUE);
+ }
+
+ @Override
+ public void start() throws ServiceException {
+ if (!cleanupThread.isAlive()) {
+ cleanupThread.setName("CleanUp Socket Thread");
+ cleanupThread.setDaemon(true);
+ cleanupThread.start();
+ }
+ }
+
+ @Override
+ public void stop() throws ServiceException {
+ if (cleanupThread.isAlive()) {
+ cleanupThread.setActive(false);
+ try {
+ cleanupThread.join(cleanupTimeout);
+ } catch (final InterruptedException e) {
+ // no-op
+ }
+ cleanupThread.interrupt();
+ }
+ }
+
+ @Override
public void service(final InputStream in, final OutputStream out) throws ServiceException, IOException {
}
@@ -224,6 +260,10 @@ public class ServicePool extends ServerS
} finally {
//Ensure delegated socket is closed here
+ final Runnable closeTask = new CloseTask(socket);
+ if (!cleanupThread.isAlive() || !cleanupThread.add(closeTask)) {
+ closeTask.run();
+ }
try {
if (socket != null) {
@@ -346,4 +386,83 @@ public class ServicePool extends ServerS
}
}
}
+
+ private class CloseTask implements Runnable {
+ private final Socket socket;
+
+ public CloseTask(final Socket socket) {
+ this.socket = socket;
+ }
+
+ @Override
+ public void run() {
+
+ int count = 0;
+ try {
+ final InputStream in = socket.getInputStream(); // should throw an exception is socket is closed
+ while (in.read() != -1) {
+ // wait client cuts the connection
+ count++;
+ if (count > maxResponseSize && maxResponseSize >= 0) {
+ break;
+ }
+ }
+ in.close();
+ } catch (final Throwable e) {
+ //Ignore
+ }
+
+ try {
+ socket.getOutputStream().close();
+ } catch (final Throwable e) {
+ //Ignore
+ }
+
+ try {
+ socket.close();
+ } catch (Throwable e) {
+ log.error("Encountered problem while closing connection with client: " + e.getMessage());
+ }
+ }
+ }
+
+ private static class CleanUpThread extends Thread {
+ private final AtomicBoolean active = new AtomicBoolean(true);
+ private final BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
+
+ @Override
+ public void run() {
+ while (active.get()) {
+ try {
+ queue.take().run();
+ } catch (final InterruptedException e) {
+ // no-op
+ }
+ }
+ }
+
+ public void setActive(final boolean value) {
+ active.set(value);
+
+ // close all
+ if (!value) {
+ for (final Runnable r : queue) {
+ r.run();
+ }
+ }
+
+ queue.add(new NoOpRunnable()); // if queue was empty simply force the loop to stop
+ }
+
+ public boolean add(final Runnable run) {
+ return queue.add(run);
+ }
+ }
+
+ private static class NoOpRunnable implements Runnable {
+ @Override
+ public void run() {
+ // no-op
+ }
+ }
}