Add TcpChannel to unify Transport implementations

core/src/main/java/org/elasticsearch/action/ActionListener.java

@@ -69,6 +69,28 @@ public void onFailure(Exception e) {
         };
     }
 
+    /**
+     * Creates a listener that listens for a response (or failure) and executes the
+     * corresponding runnable when the response (or failure) is received.
+     *
+     * @param runnable the runnable that will be called in event of success or failure
+     * @param <Response> the type of the response
+     * @return a listener that listens for responses and invokes the runnable when received
+     */
+    static <Response> ActionListener<Response> wrap(Runnable runnable) {
+        return new ActionListener<Response>() {
+            @Override
+            public void onResponse(Response response) {
+                runnable.run();
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                runnable.run();
+            }
+        };
+    }
+
     /**
      * Notifies every given listener with the response passed to {@link #onResponse(Object)}. If a listener itself throws an exception
      * the exception is forwarded to {@link #onFailure(Exception)}. If in turn {@link #onFailure(Exception)} fails all remaining

core/src/main/java/org/elasticsearch/action/support/ListenerExecutionContext.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.action.support;
+
+import org.elasticsearch.action.ActionListener;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class ListenerExecutionContext<V> implements ActionListener<V> {
+
+    private static Object NULL_VALUE = new Object();
+
+    private final ConcurrentLinkedQueue<ActionListener<V>> listeners = new ConcurrentLinkedQueue<>();
+    private final AtomicReference<Object> result = new AtomicReference<>(null);
+
+    @Override
+    public void onResponse(V value) {
+        if (result.compareAndSet(null, value != null ? value : NULL_VALUE)) {
+            ActionListener<V> listener;
+            while ((listener = listeners.poll()) != null) {
+                listener.onResponse(value);
+            }
+        }
+    }
+
+    @Override
+    public void onFailure(Exception e) {
+        if (e == null) {
+            throw new IllegalArgumentException("Exception cannot be null");
+        }
+        if (result.compareAndSet(null, e)) {
+            ActionListener<V> listener;
+            while ((listener = listeners.poll()) != null) {
+                listener.onFailure(e);
+            }
+        }
+    }
+
+    public boolean isDone() {
+        return result.get() != null;
+    }
+
+    public void addListener(ActionListener<V> listener) {
+        internalAddListener(listener);
+    }
+
+    @SuppressWarnings("unchecked")
+    private void internalAddListener(ActionListener<V> listener) {
+        listeners.offer(listener);
+
+        Object result = this.result.get();
+        if (result != null) {
+            if (listeners.remove(listener)) {
+                if (result instanceof Exception) {
+                    listener.onFailure((Exception) result);
+                } else if (result == NULL_VALUE) {
+                    listener.onResponse(null);
+                } else {
+                    listener.onResponse((V) result);
+                }
+            }
+        }
+    }
+}

core/src/main/java/org/elasticsearch/transport/ConnectionProfile.java

@@ -208,12 +208,12 @@ private ConnectionTypeHandle(int offset, int length, Set<TransportRequestOptions
          * Returns one of the channels out configured for this handle. The channel is selected in a round-robin
          * fashion.
          */
-        <T> T getChannel(T[] channels) {
+        <T> T getChannel(List<T> channels) {
             if (length == 0) {
                 throw new IllegalStateException("can't select channel size is 0 for types: " + types);
             }
-            assert channels.length >= offset + length : "illegal size: " + channels.length + " expected >= " + (offset + length);
-            return channels[offset + Math.floorMod(counter.incrementAndGet(), length)];
+            assert channels.size() >= offset + length : "illegal size: " + channels.size() + " expected >= " + (offset + length);
+            return channels.get(offset + Math.floorMod(counter.incrementAndGet(), length));
         }
 
         /**
@@ -223,5 +223,4 @@ Set<TransportRequestOptions.Type> getTypes() {
             return types;
         }
     }
-
 }

core/src/main/java/org/elasticsearch/transport/TcpChannel.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.transport;
+
+import org.elasticsearch.action.ActionListener;
+
+import java.io.IOException;
+
+public interface TcpChannel<C extends TcpChannel<C>> {
+
+    void closeAsync();
+
+    void addCloseListener(ActionListener<C> listener);
+
+    void setSoLinger(int value) throws IOException;
+
+    boolean isOpen();
+}

core/src/main/java/org/elasticsearch/transport/TcpChannelUtils.java

@@ -0,0 +1,137 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.transport;
+
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ListenableActionFuture;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.unit.TimeValue;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class TcpChannelUtils {
+
+    public static <C extends TcpChannel<C>> void closeChannel(C channel, boolean blocking) {
+        if (channel.isOpen()) {
+            if (blocking) {
+                PlainActionFuture<C> closeFuture = PlainActionFuture.newFuture();
+                channel.addCloseListener(closeFuture);
+                channel.closeAsync();
+                blockOnFutures(Collections.singletonList(closeFuture));
+            } else {
+                channel.closeAsync();
+            }
+        }
+    }
+
+    public static <C extends TcpChannel<C>> void closeChannels(List<C> channels, boolean blocking) {
+        if (blocking) {
+            ArrayList<ActionFuture<C>> futures = new ArrayList<>(channels.size());
+            for (final C channel : channels) {
+                if (channel.isOpen()) {
+                    PlainActionFuture<C> closeFuture = PlainActionFuture.newFuture();
+                    channel.addCloseListener(closeFuture);
+                    channel.closeAsync();
+                    futures.add(closeFuture);
+                }
+            }
+            blockOnFutures(futures);
+        } else {
+            channels.forEach(c -> {
+                if (c.isOpen()) {
+                    c.closeAsync();
+                }
+            });
+        }
+    }
+
+    public static <C extends TcpChannel<C>> void closeServerChannels(String profile, List<C> channels, Logger logger) {
+        ArrayList<ActionFuture<C>> futures = new ArrayList<>(channels.size());
+        for (final C channel : channels) {
+            if (channel.isOpen()) {
+                PlainActionFuture<C> closeFuture = PlainActionFuture.newFuture();
+                channel.addCloseListener(ActionListener.wrap(c -> {
+                    },
+                    e -> logger.warn(() -> new ParameterizedMessage("Error closing serverChannel for profile [{}]", profile), e)));
+                channel.addCloseListener(closeFuture);
+                channel.closeAsync();
+                futures.add(closeFuture);
+            }
+        }
+
+        blockOnFutures(futures);
+    }
+
+    public static <C extends TcpChannel<C>> void finishConnection(DiscoveryNode discoveryNode, List<ActionFuture<C>> connectionFutures,
+                                                                  TimeValue connectTimeout) throws ConnectTransportException {
+        Exception connectionException = null;
+        boolean allConnected = true;
+
+        for (ActionFuture<C> connectionFuture : connectionFutures) {
+            try {
+                connectionFuture.get(connectTimeout.getMillis(), TimeUnit.MILLISECONDS);
+            } catch (TimeoutException e) {
+                allConnected = false;
+                break;
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IllegalStateException(e);
+            } catch (ExecutionException e) {
+                allConnected = false;
+                connectionException = (Exception) e.getCause();
+                break;
+            }
+        }
+
+        if (allConnected == false) {
+            if (connectionException == null) {
+                throw new ConnectTransportException(discoveryNode, "connect_timeout[" + connectTimeout + "]");
+            } else {
+                throw new ConnectTransportException(discoveryNode, "connect_exception", connectionException);
+            }
+        }
+    }
+
+
+    private static <C extends TcpChannel<C>> void blockOnFutures(List<ActionFuture<C>> futures) {
+        for (ActionFuture<C> future : futures) {
+            try {
+                future.get();
+            } catch (ExecutionException e) {
+                // Ignore as we already attached a listener to log
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IllegalStateException("Future got interrupted", e);
+            }
+        }
+    }
+}