Refactor ConcurrentPool so that it becomes agnostic of the item initi…

…alization concept

JAVA-3927

driver-core/src/main/com/mongodb/internal/connection/ConcurrentPool.java

@@ -16,12 +16,15 @@
 
 package com.mongodb.internal.connection;
 
+import com.mongodb.Function;
 import com.mongodb.MongoInternalException;
 import com.mongodb.MongoInterruptedException;
 import com.mongodb.MongoTimeoutException;
 import com.mongodb.internal.connection.ConcurrentLinkedDeque.RemovalReportingIterator;
+import com.mongodb.lang.Nullable;
 
 import java.util.Iterator;
+import java.util.Optional;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
@@ -59,7 +62,7 @@ public enum Prune {
      * @param <T>
      */
     public interface ItemFactory<T> {
-        T create(boolean initialize);
+        T create();
 
         void close(T t);
 
@@ -143,7 +146,8 @@ public T get(final long timeout, final TimeUnit timeUnit) {
 
         T t = available.pollLast();
         if (t == null) {
-            t = createNewAndReleasePermitIfFailure(false);
+            t = createNewAndReleasePermitIfFailure(null);
+            assert t != null : "new item must not be null if postCreate is null";
         }
 
         return t;
@@ -167,25 +171,53 @@ public void prune() {
         }
     }
 
-    public void ensureMinSize(final int minSize, final boolean initialize) {
+    /**
+     * Try to populate this pool with items so that {@link #getCount()} is not smaller than {@code minSize}.
+     * The {@code postCreate} action returning an {@linkplain Optional#isEmpty() empty} result causes this method to return.
+     *
+     * @param postCreate A transforming action applied to non-{@code null} new items.
+     *                   This action may return an item as is, may modify the item,
+     *                   or return a (potentially {@linkplain Optional#isEmpty() empty}) reference to different object / throw an
+     *                   {@link Exception}. In the latter case the action must release
+     *                   resources associated with the item if they cannot be releases via the returned reference.
+     */
+    public void ensureMinSize(final int minSize, @Nullable final Function<? super T, ? extends Optional<? extends T>> postCreate) {
         while (getCount() < minSize) {
-            if (!acquirePermit(10, TimeUnit.MILLISECONDS)) {
+            if (!acquirePermit(0, TimeUnit.MILLISECONDS)) {
                 break;
             }
-            release(createNewAndReleasePermitIfFailure(initialize));
+            T newItem = createNewAndReleasePermitIfFailure(postCreate);
+            if (newItem == null) {
+                break;
+            } else {
+                release(newItem);
+            }
         }
     }
 
-    private T createNewAndReleasePermitIfFailure(final boolean initialize) {
+    /**
+     * If this method returns {@code null} or throws an {@link Exception}, then before doing so it releases a permit.
+     *
+     * @param postCreate See {@link #ensureMinSize(int, Function)}.
+     * @return Either a {@linkplain ItemFactory#create() new} non-{@code null} item if {@code postCreate} is {@code null},
+     * or the result of {@linkplain Function#apply(Object) applying} {@code postCreate} to the new item
+     * (this result may be {@linkplain Optional#isEmpty() empty}, in which case the method returns {@code null}).
+     */
+    @Nullable
+    private T createNewAndReleasePermitIfFailure(@Nullable final Function<? super T, ? extends Optional<? extends T>> postCreate) {
+        boolean failure = true;
         try {
-            T newMember = itemFactory.create(initialize);
+            T newMember = itemFactory.create();
             if (newMember == null) {
                 throw new MongoInternalException("The factory for the pool created a null item");
             }
+            newMember = postCreate == null ? newMember : postCreate.apply(newMember).orElse(null);
+            failure = newMember == null;
             return newMember;
-        } catch (RuntimeException e) {
-            permits.release();
-            throw e;
+        } finally {
+            if (failure) {
+                permits.release();
+            }
         }
     }
 

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -45,6 +45,7 @@
 import org.bson.codecs.Decoder;
 
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -320,7 +321,13 @@ public synchronized void run() {
                             if (LOGGER.isDebugEnabled()) {
                                 LOGGER.debug(format("Ensuring minimum pooled connections to %s", serverId.getAddress()));
                             }
-                            pool.ensureMinSize(settings.getMinSize(), true);
+                            pool.ensureMinSize(settings.getMinSize(), newConnection -> {
+                                if (!newConnection.opened()) {
+                                    newConnection.open();
+                                    connectionPoolListener.connectionReady(new ConnectionReadyEvent(getId(newConnection)));
+                                }
+                                return Optional.of(newConnection);
+                            });
                         }
                     } catch (MongoInterruptedException e) {
                         // don't log interruptions due to the shutdownNow call on the ExecutorService
@@ -580,16 +587,11 @@ private class UsageTrackingInternalConnectionItemFactory implements ConcurrentPo
         }
 
         @Override
-        public UsageTrackingInternalConnection create(final boolean initialize) {
+        public UsageTrackingInternalConnection create() {
             UsageTrackingInternalConnection internalConnection =
             new UsageTrackingInternalConnection(internalConnectionFactory.create(serverId), generation.get());
             ConnectionId id = getId(internalConnection);
             connectionCreated(connectionPoolListener, id);
-            if (initialize) {
-                internalConnection.open();
-                connectionPoolListener.connectionReady(new ConnectionReadyEvent(id));
-            }
-
             return internalConnection;
         }
 

driver-core/src/main/com/mongodb/internal/connection/PowerOfTwoBufferPool.java

@@ -55,7 +55,7 @@ public PowerOfTwoBufferPool(final int highestPowerOfTwo) {
             powerOfTwoToPoolMap.put(i, new ConcurrentPool<ByteBuffer>(Integer.MAX_VALUE,
                                                                          new ConcurrentPool.ItemFactory<ByteBuffer>() {
                                                                              @Override
-                                                                             public ByteBuffer create(final boolean initialize) {
+                                                                             public ByteBuffer create() {
                                                                                  return createNew(size);
                                                                              }
 

driver-core/src/main/com/mongodb/internal/session/ServerSessionPool.java

@@ -241,7 +241,7 @@ public boolean isMarkedDirty() {
 
     private final class ServerSessionItemFactory implements ConcurrentPool.ItemFactory<ServerSessionImpl> {
         @Override
-        public ServerSessionImpl create(final boolean initialize) {
+        public ServerSessionImpl create() {
             return new ServerSessionImpl(createNewServerSessionIdentifier());
         }
 

driver-core/src/test/unit/com/mongodb/internal/connection/ConcurrentPoolTest.java

@@ -16,11 +16,13 @@
 
 package com.mongodb.internal.connection;
 
+import com.mongodb.Function;
 import com.mongodb.MongoException;
 import com.mongodb.MongoTimeoutException;
 import org.junit.Test;
 
 import java.io.Closeable;
+import java.util.Optional;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.junit.Assert.assertEquals;
@@ -141,36 +143,36 @@ public void testCloseItemOnReleaseAfterPoolClosed() {
     public void testEnsureMinSize() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
 
-        pool.ensureMinSize(0, false);
+        pool.ensureMinSize(0, null);
         assertEquals(0, pool.getAvailableCount());
 
-        pool.ensureMinSize(1, false);
+        pool.ensureMinSize(1, null);
         assertEquals(1, pool.getAvailableCount());
 
-        pool.ensureMinSize(1, false);
+        pool.ensureMinSize(1, null);
         assertEquals(1, pool.getAvailableCount());
 
         pool.get();
-        pool.ensureMinSize(1, false);
+        pool.ensureMinSize(1, null);
         assertEquals(0, pool.getAvailableCount());
 
-        pool.ensureMinSize(4, false);
+        pool.ensureMinSize(4, null);
         assertEquals(3, pool.getAvailableCount());
     }
 
     @Test
     public void whenEnsuringMinSizeShouldNotInitializePooledItemIfNotRequested() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
 
-        pool.ensureMinSize(1, false);
+        pool.ensureMinSize(1, null);
         assertFalse(pool.get().isInitialized());
     }
 
     @Test
     public void whenEnsuringMinSizeShouldInitializePooledItemIfRequested() {
         pool = new ConcurrentPool<TestCloseable>(3, new TestItemFactory());
 
-        pool.ensureMinSize(1, true);
+        pool.ensureMinSize(1, TestCloseable.initializeAction);
         assertTrue(pool.get().isInitialized());
     }
 
@@ -179,7 +181,7 @@ public void testThatEnsuringMinSizeReleasesPermitIfCreateFails() {
         pool = new ConcurrentPool<TestCloseable>(1, new TestItemFactory(true));
 
         try {
-            pool.ensureMinSize(1, true);
+            pool.ensureMinSize(1, TestCloseable.initializeAction);
             fail();
         } catch (MongoException e) {
             // expected
@@ -232,11 +234,11 @@ class TestItemFactory implements ConcurrentPool.ItemFactory<TestCloseable> {
         }
 
         @Override
-        public TestCloseable create(final boolean initialize) {
+        public TestCloseable create() {
             if (shouldThrowOnCreate) {
                 throw new MongoException("This is a journey");
             }
-            return new TestCloseable(initialize);
+            return new TestCloseable();
         }
 
         @Override
@@ -251,12 +253,16 @@ public ConcurrentPool.Prune shouldPrune(final TestCloseable testCloseable) {
     }
 
     static class TestCloseable implements Closeable {
+        private static final Function<TestCloseable, Optional<TestCloseable>> initializeAction = connection -> {
+            connection.initialized = true;
+            return Optional.of(connection);
+        };
+
         private boolean closed;
         private ConcurrentPool.Prune shouldPrune;
-        private final boolean initialized;
+        private boolean initialized;
 
-        TestCloseable(final boolean initialize) {
-            this.initialized = initialize;
+        TestCloseable() {
         }
 
         @Override