LOGBACK-945: extended ServerSocketReceiver and AbstractServerSocketAppender have their respective own thread pools for managing client connections

logback-classic/src/main/java/ch/qos/logback/classic/net/server/ServerSocketReceiver.java

@@ -18,19 +18,24 @@
 import java.net.ServerSocket;
 import java.net.UnknownHostException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 import javax.net.ServerSocketFactory;
 
 import ch.qos.logback.classic.net.ReceiverBase;
+import ch.qos.logback.core.CoreConstants;
 import ch.qos.logback.core.net.AbstractSocketAppender;
 import ch.qos.logback.core.net.server.ServerListener;
 import ch.qos.logback.core.net.server.ServerRunner;
-import ch.qos.logback.core.util.CloseUtil;
 
 /**
  * A logging socket server that is configurable using Joran.
  *
  * @author Carl Harris
+ * @author Sebastian Gröbler
  */
 public class ServerSocketReceiver extends ReceiverBase {
 
@@ -44,8 +49,11 @@ public class ServerSocketReceiver extends ReceiverBase {
 
   private String address;
 
-  private ServerSocket serverSocket;
   private ServerRunner runner;
+
+  private int corePoolSize = CoreConstants.CORE_POOL_SIZE;
+  private int maxPoolSize = CoreConstants.MAX_POOL_SIZE;
+  protected ExecutorService connectionPoolExecutorService;
 
   /**
    * Starts the server.
@@ -58,13 +66,12 @@ protected boolean shouldStart() {
       ServerListener<RemoteAppenderClient> listener = 
           createServerListener(serverSocket);
 
-      runner = createServerRunner(listener, getContext().getExecutorService());
+      runner = createServerRunner(listener, getConnectionPoolExecutorService());
       runner.setContext(getContext());
       return true;
     }
     catch (Exception ex) {
       addError("server startup error: " + ex, ex);
-      CloseUtil.closeQuietly(serverSocket);
       return false;
     }
   }
@@ -95,7 +102,29 @@ protected void onStop() {
     }
     catch (IOException ex) {
       addError("server shutdown error: " + ex, ex);
+    } finally {
+      shutDownExecutorService();
+    }
+  }
+
+  private synchronized void shutDownExecutorService() {
+    connectionPoolExecutorService.shutdownNow();
+    connectionPoolExecutorService = null;
+  }
+
+  private ExecutorService getConnectionPoolExecutorService() {
+    if (connectionPoolExecutorService == null) {
+      synchronized (this) {
+        if (connectionPoolExecutorService == null) {
+          connectionPoolExecutorService = new ThreadPoolExecutor(
+                  getCorePoolSize(),
+                  getMaxPoolSize(),
+                  0L, TimeUnit.MILLISECONDS,
+                  new SynchronousQueue<Runnable>());
+        }
+      }
     }
+    return connectionPoolExecutorService;
   }
 
   /**
@@ -175,4 +204,37 @@ public void setAddress(String address) {
     this.address = address;
   }
 
+  /**
+   * Gets the core pool size for the socket client connection pool.
+   * The default value is {@link CoreConstants#CORE_POOL_SIZE}.
+   * @return the core pool size
+   */
+  public int getCorePoolSize() {
+    return corePoolSize;
+  }
+
+  /**
+   * Sets the core number of threads for the socket client connection pool.
+   * @param corePoolSize the core pool size
+   */
+  public void setCorePoolSize(int corePoolSize) {
+    this.corePoolSize = corePoolSize;
+  }
+
+  /**
+   * Gets the maximum pool size for the socket client connection pool.
+   * The default value is {@link CoreConstants#MAX_POOL_SIZE}.
+   * @return the maximum pool size
+   */
+  public int getMaxPoolSize() {
+    return maxPoolSize;
+  }
+
+  /**
+   * Sets the maximum allowed number of threads for the socket client connection pool.
+   * @param maxPoolSize the maximum pool size
+   */
+  public void setMaxPoolSize(int maxPoolSize) {
+    this.maxPoolSize = maxPoolSize;
+  }
 }

logback-classic/src/test/java/ch/qos/logback/classic/net/server/ConnectionDropCountingServerSocketAppender.java

@@ -0,0 +1,139 @@
+/**
+ * Logback: the reliable, generic, fast and flexible logging framework.
+ * Copyright (C) 1999-2014, QOS.ch. All rights reserved.
+ *
+ * This program and the accompanying materials are dual-licensed under
+ * either the terms of the Eclipse Public License v1.0 as published by
+ * the Eclipse Foundation
+ *
+ *   or (per the licensee's choosing)
+ *
+ * under the terms of the GNU Lesser General Public License version 2.1
+ * as published by the Free Software Foundation.
+ */
+
+package ch.qos.logback.classic.net.server;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.net.ServerSocket;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+
+import ch.qos.logback.classic.net.LoggingEventPreSerializationTransformer;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.Context;
+import ch.qos.logback.core.net.server.AbstractServerSocketAppender;
+import ch.qos.logback.core.net.server.RemoteReceiverClient;
+import ch.qos.logback.core.net.server.ServerListener;
+import ch.qos.logback.core.spi.PreSerializationTransformer;
+import ch.qos.logback.core.status.Status;
+
+/**
+ * Implementation of {@link AbstractServerSocketAppender} which allows to observe how many client connections have been closed.
+ *
+ * @author Sebastian Gröbler
+ */
+public class ConnectionDropCountingServerSocketAppender extends AbstractServerSocketAppender<ILoggingEvent> {
+
+    private final CountDownLatch closeCounter;
+    private PreSerializationTransformer<ILoggingEvent> pst = new LoggingEventPreSerializationTransformer();
+
+    public ConnectionDropCountingServerSocketAppender(final CountDownLatch closeCounter) {
+        this.closeCounter = closeCounter;
+    }
+
+    @Override
+    protected void postProcessEvent(final ILoggingEvent event) {
+        event.prepareForDeferredProcessing();
+    }
+
+    @Override
+    protected PreSerializationTransformer<ILoggingEvent> getPST() {
+        return pst;
+    }
+
+    @Override
+    protected ServerListener<RemoteReceiverClient> createServerListener(final ServerSocket socket) {
+        return new DropCountingListener(super.createServerListener(socket));
+    }
+
+    private class DropCountingListener implements ServerListener<RemoteReceiverClient> {
+
+        private final ServerListener<RemoteReceiverClient> decoratedListener;
+
+        private DropCountingListener(final ServerListener<RemoteReceiverClient> decoratedListener) {
+            this.decoratedListener = decoratedListener;
+        }
+
+        public RemoteReceiverClient acceptClient() throws IOException, InterruptedException {
+            return new DropCountingClient(decoratedListener.acceptClient());
+        }
+
+        public void close() {
+            decoratedListener.close();
+        }
+    }
+
+    private class DropCountingClient implements RemoteReceiverClient {
+
+        private final RemoteReceiverClient decoratedClient;
+
+        private DropCountingClient(final RemoteReceiverClient decoratedClient) {
+            this.decoratedClient = decoratedClient;
+        }
+
+        public void close() {
+            closeCounter.countDown();
+            decoratedClient.close();
+        }
+
+        public void run() {
+            decoratedClient.run();
+        }
+
+        public void setQueue(final BlockingQueue<Serializable> queue) {
+            decoratedClient.setQueue(queue);
+        }
+
+        public boolean offer(final Serializable event) {
+            return decoratedClient.offer(event);
+        }
+
+        public void setContext(final Context context) {
+            decoratedClient.setContext(context);
+        }
+
+        public Context getContext() {
+            return decoratedClient.getContext();
+        }
+
+        public void addStatus(final Status status) {
+            decoratedClient.addStatus(status);
+        }
+
+        public void addInfo(final String msg) {
+            decoratedClient.addInfo(msg);
+        }
+
+        public void addInfo(final String msg, final Throwable ex) {
+            decoratedClient.addInfo(msg, ex);
+        }
+
+        public void addWarn(final String msg) {
+            decoratedClient.addWarn(msg);
+        }
+
+        public void addWarn(final String msg, final Throwable ex) {
+            decoratedClient.addWarn(msg, ex);
+        }
+
+        public void addError(final String msg) {
+            decoratedClient.addError(msg);
+        }
+
+        public void addError(final String msg, final Throwable ex) {
+            decoratedClient.addError(msg, ex);
+        }
+    }
+}

logback-classic/src/test/java/ch/qos/logback/classic/net/server/ConnectionDropCountingServerSocketReceiver.java

@@ -0,0 +1,79 @@
+/**
+ * Logback: the reliable, generic, fast and flexible logging framework.
+ * Copyright (C) 1999-2014, QOS.ch. All rights reserved.
+ *
+ * This program and the accompanying materials are dual-licensed under
+ * either the terms of the Eclipse Public License v1.0 as published by
+ * the Eclipse Foundation
+ *
+ *   or (per the licensee's choosing)
+ *
+ * under the terms of the GNU Lesser General Public License version 2.1
+ * as published by the Free Software Foundation.
+ */
+package ch.qos.logback.classic.net.server;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.concurrent.CountDownLatch;
+
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.core.net.server.ServerListener;
+
+/**
+ * Extension of {@link ServerSocketReceiver} which allows to observe how many client connections have been closed.
+ *
+ * @author Sebastian Gröbler
+ */
+public class ConnectionDropCountingServerSocketReceiver extends ServerSocketReceiver {
+
+    protected final CountDownLatch closeCounter;
+
+    public ConnectionDropCountingServerSocketReceiver(final CountDownLatch closeCounter) {
+        this.closeCounter = closeCounter;
+    }
+
+    @Override
+    protected ServerListener<RemoteAppenderClient> createServerListener(final ServerSocket socket) {
+        return new DropCountingListener(super.createServerListener(socket));
+    }
+
+    private class DropCountingListener implements ServerListener<RemoteAppenderClient> {
+
+        private final ServerListener<RemoteAppenderClient> decoratedListener;
+
+        private DropCountingListener(final ServerListener<RemoteAppenderClient> decoratedListener) {
+            this.decoratedListener = decoratedListener;
+        }
+
+        public RemoteAppenderClient acceptClient() throws IOException, InterruptedException {
+            return new DropCountingClient(decoratedListener.acceptClient());
+        }
+
+        public void close() {
+            decoratedListener.close();
+        }
+    }
+
+    private class DropCountingClient implements RemoteAppenderClient {
+
+        private final RemoteAppenderClient decoratedClient;
+
+        private DropCountingClient(final RemoteAppenderClient decoratedClient) {
+            this.decoratedClient = decoratedClient;
+        }
+
+        public void setLoggerContext(final LoggerContext lc) {
+            decoratedClient.setLoggerContext(lc);
+        }
+
+        public void close() {
+            closeCounter.countDown();
+            decoratedClient.close();
+        }
+
+        public void run() {
+            decoratedClient.run();
+        }
+    }
+}