Resolved issue: NETTY-85 (Global Channel registry)

* Initial implementation

src/main/java/org/jboss/netty/group/ChannelGroup.java

@@ -0,0 +1,46 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import java.net.SocketAddress;
+import java.util.Set;
+import java.util.UUID;
+
+import org.jboss.netty.channel.Channel;
+
+/**
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ * @version $Rev$, $Date$
+ */
+public interface ChannelGroup extends Set<Channel>, Comparable<ChannelGroup> {
+    String getName();
+    Channel find(UUID id);
+    ChannelGroupFuture setInterestOps(int interestOps);
+    ChannelGroupFuture setReadable(boolean readable);
+    ChannelGroupFuture write(Object message);
+    ChannelGroupFuture write(Object message, SocketAddress remoteAddress);
+    ChannelGroupFuture disconnect();
+    ChannelGroupFuture unbind();
+    ChannelGroupFuture close();
+}

src/main/java/org/jboss/netty/group/ChannelGroupFactory.java

@@ -0,0 +1,57 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+
+/**
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ * @version $Rev$, $Date$
+ */
+public class ChannelGroupFactory {
+
+    private static final ConcurrentMap<String, ChannelGroup> groups =
+        new ConcurrentHashMap<String, ChannelGroup>();
+
+    public static ChannelGroup getGroup(Class<?> groupType) {
+        return getGroup(groupType.getName());
+    }
+
+    public static ChannelGroup getGroup(String groupName) {
+        ChannelGroup g = groups.get(groupName);
+        if (g != null) {
+            return g;
+        }
+
+        g = new DefaultChannelGroup(groupName);
+        g = groups.putIfAbsent(groupName, g);
+        return g;
+    }
+
+    private ChannelGroupFactory() {
+        super();
+    }
+}

src/main/java/org/jboss/netty/group/ChannelGroupFuture.java

@@ -0,0 +1,133 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+
+/**
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ * @version $Rev$, $Date$
+ */
+public interface ChannelGroupFuture extends Iterable<ChannelFuture>{
+
+    ChannelGroup getGroup();
+
+    ChannelFuture find(UUID channelId);
+    ChannelFuture find(Channel channel);
+
+    /**
+     * Returns {@code true} if and only if this future is
+     * complete, regardless of whether the operation was successful, failed,
+     * or canceled.
+     */
+    boolean isDone();
+
+    boolean isCompleteSuccess();
+
+    boolean isPartialSuccess();
+
+    boolean isCompleteFailure();
+
+    boolean isPartialFailure();
+
+    /**
+     * Adds the specified listener to this future.  The
+     * specified listener is notified when this future is
+     * {@linkplain #isDone() done}.  If this future is already
+     * completed, the specified listener is notified immediately.
+     */
+    void addListener(ChannelGroupFutureListener listener);
+
+    /**
+     * Removes the specified listener from this future.
+     * The specified listener is no longer notified when this
+     * future is {@linkplain #isDone() done}.  If this
+     * future is already completed, this method has no effect
+     * and returns silently.
+     */
+    void removeListener(ChannelGroupFutureListener listener);
+
+    /**
+     * Waits for this future to be completed.
+     *
+     * @throws InterruptedException
+     *         if the current thread was interrupted
+     */
+    ChannelGroupFuture await() throws InterruptedException;
+
+    /**
+     * Waits for this future to be completed without
+     * interruption.  This method catches an {@link InterruptedException} and
+     * discards it silently.
+     */
+    ChannelGroupFuture awaitUninterruptibly();
+
+    /**
+     * Waits for this future to be completed within the
+     * specified time limit.
+     *
+     * @return {@code true} if and only if the future was completed within
+     *         the specified time limit
+     *
+     * @throws InterruptedException
+     *         if the current thread was interrupted
+     */
+    boolean await(long timeout, TimeUnit unit) throws InterruptedException;
+
+    /**
+     * Waits for this future to be completed within the
+     * specified time limit.
+     *
+     * @return {@code true} if and only if the future was completed within
+     *         the specified time limit
+     *
+     * @throws InterruptedException
+     *         if the current thread was interrupted
+     */
+    boolean await(long timeoutMillis) throws InterruptedException;
+
+    /**
+     * Waits for this future to be completed within the
+     * specified time limit without interruption.  This method catches an
+     * {@link InterruptedException} and discards it silently.
+     *
+     * @return {@code true} if and only if the future was completed within
+     *         the specified time limit
+     */
+    boolean awaitUninterruptibly(long timeout, TimeUnit unit);
+
+    /**
+     * Waits for this future to be completed within the
+     * specified time limit without interruption.  This method catches an
+     * {@link InterruptedException} and discards it silently.
+     *
+     * @return {@code true} if and only if the future was completed within
+     *         the specified time limit
+     */
+    boolean awaitUninterruptibly(long timeoutMillis);
+}

src/main/java/org/jboss/netty/group/ChannelGroupFutureListener.java

@@ -0,0 +1,47 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import java.util.EventListener;
+
+/**
+ * Listens to the result of a {@link ChannelGroupFuture}.  The result of the
+ * asynchronous {@link ChannelGroup} I/O operation is notified once this
+ * listener is added by calling {@link ChannelGroupFuture#addListener(ChannelGroupFutureListener)}.
+ *
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ *
+ * @version $Rev$, $Date$
+ */
+public interface ChannelGroupFutureListener extends EventListener {
+
+    /**
+     * Invoked when the I/O operation associated with the
+     * {@link ChannelGroupFuture} has been completed.
+     *
+     * @param future  The source {@link ChannelGroupFuture} which called this
+     *                callback.
+     */
+    void operationComplete(ChannelGroupFuture future) throws Exception;
+}

src/main/java/org/jboss/netty/group/DefaultChannelGroup.java

@@ -0,0 +1,221 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import java.net.SocketAddress;
+import java.util.AbstractSet;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+
+/**
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ * @version $Rev$, $Date$
+ */
+public class DefaultChannelGroup extends AbstractSet<Channel> implements ChannelGroup {
+
+    private final String name;
+    private final ConcurrentMap<UUID, Channel> channels = new ConcurrentHashMap<UUID, Channel>();
+    private final ChannelFutureListener remover = new ChannelFutureListener() {
+        public void operationComplete(ChannelFuture future) throws Exception {
+            remove(future.getChannel());
+        }
+    };
+
+    public DefaultChannelGroup(String name) {
+        if (name == null) {
+            throw new NullPointerException("name");
+        }
+        this.name = name;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return channels.isEmpty();
+    }
+
+    @Override
+    public int size() {
+        return channels.size();
+    }
+
+    public Channel find(UUID id) {
+        return channels.get(id);
+    }
+
+    @Override
+    public boolean contains(Object o) {
+        if (o instanceof UUID) {
+            return channels.containsKey(o);
+        } else if (o instanceof Channel) {
+            return channels.containsKey(((Channel) o).getId());
+        } else {
+            return false;
+        }
+    }
+
+    @Override
+    public boolean add(Channel channel) {
+        boolean added = channels.putIfAbsent(channel.getId(), channel) == null;
+        if (added) {
+            channel.getCloseFuture().addListener(remover);
+        }
+        return added;
+    }
+
+    @Override
+    public boolean remove(Object o) {
+        Channel c = null;
+        if (o instanceof UUID) {
+            c = channels.remove(o);
+        } else if (o instanceof Channel) {
+            c = channels.remove(((Channel) o).getId());
+        }
+
+        if (c == null) {
+            return false;
+        }
+
+        c.getCloseFuture().removeListener(remover);
+        return true;
+    }
+
+    @Override
+    public void clear() {
+        channels.clear();
+    }
+
+    @Override
+    public Iterator<Channel> iterator() {
+        return channels.values().iterator();
+    }
+
+    @Override
+    public Object[] toArray() {
+        return channels.values().toArray();
+    }
+
+    @Override
+    public <T> T[] toArray(T[] a) {
+        return channels.values().toArray(a);
+    }
+
+    public ChannelGroupFuture close() {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.close());
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture disconnect() {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.disconnect());
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture setInterestOps(int interestOps) {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.setInterestOps(interestOps));
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture setReadable(boolean readable) {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.setReadable(readable));
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture unbind() {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.unbind());
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture write(Object message) {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.write(message));
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    public ChannelGroupFuture write(Object message, SocketAddress remoteAddress) {
+        Map<UUID, ChannelFuture> futures =
+            new HashMap<UUID, ChannelFuture>(size());
+        for (Channel c: this) {
+            futures.put(c.getId(), c.write(message, remoteAddress));
+        }
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    @Override
+    public int hashCode() {
+        return getName().hashCode();
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof ChannelGroup)) {
+            return false;
+        }
+
+        ChannelGroup that = (ChannelGroup) o;
+        return getName().equals(that.getName());
+    }
+
+    public int compareTo(ChannelGroup o) {
+        return getName().compareTo(o.getName());
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName() + '(' + name + ')';
+    }
+}

src/main/java/org/jboss/netty/group/DefaultChannelGroupFuture.java

@@ -0,0 +1,332 @@
+/*
+ * JBoss, Home of Professional Open Source
+ *
+ * Copyright 2008, Red Hat Middleware LLC, and individual contributors
+ * by the @author tags. See the COPYRIGHT.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.netty.group;
+
+import static java.util.concurrent.TimeUnit.*;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.logging.InternalLogger;
+import org.jboss.netty.logging.InternalLoggerFactory;
+
+/**
+ * The default {@link ChannelGroupFuture} implementation.
+ *
+ * @author The Netty Project (netty-dev@lists.jboss.org)
+ * @author Trustin Lee (tlee@redhat.com)
+ *
+ * @version $Rev$, $Date$
+ */
+public class DefaultChannelGroupFuture implements ChannelGroupFuture {
+
+    private static final InternalLogger logger =
+        InternalLoggerFactory.getInstance(DefaultChannelGroupFuture.class);
+
+    private final ChannelGroup group;
+    final Map<UUID, ChannelFuture> futures;
+    private volatile ChannelGroupFutureListener firstListener;
+    private volatile List<ChannelGroupFutureListener> otherListeners;
+    private boolean done;
+    int successCount;
+    int failureCount;
+    private int waiters;
+
+    private final ChannelFutureListener childListener = new ChannelFutureListener() {
+        public void operationComplete(ChannelFuture future) throws Exception {
+            boolean success = future.isSuccess();
+            synchronized (DefaultChannelGroupFuture.this) {
+                if (success) {
+                    successCount ++;
+                } else {
+                    failureCount ++;
+                }
+            }
+
+            if (successCount + failureCount >= futures.size()) {
+                setDone();
+            }
+        }
+    };
+
+    /**
+     * Creates a new instance.
+     */
+    public DefaultChannelGroupFuture(ChannelGroup group, Collection<ChannelFuture> futures) {
+        if (group == null) {
+            throw new NullPointerException("group");
+        }
+        if (futures == null) {
+            throw new NullPointerException("futures");
+        }
+
+        this.group = group;
+
+        Map<UUID, ChannelFuture> futureMap = new HashMap<UUID, ChannelFuture>();
+        for (ChannelFuture f: futures) {
+            futureMap.put(f.getChannel().getId(), f);
+        }
+
+        this.futures = Collections.unmodifiableMap(futureMap);
+        for (ChannelFuture f: this.futures.values()) {
+            f.addListener(childListener);
+        }
+    }
+
+    DefaultChannelGroupFuture(ChannelGroup group, Map<UUID, ChannelFuture> futures) {
+        this.group = group;
+        this.futures = Collections.unmodifiableMap(futures);
+        for (ChannelFuture f: this.futures.values()) {
+            f.addListener(childListener);
+        }
+    }
+
+    public ChannelGroup getGroup() {
+        return group;
+    }
+
+    public ChannelFuture find(UUID channelId) {
+        return futures.get(channelId);
+    }
+
+    public ChannelFuture find(Channel channel) {
+        return futures.get(channel.getId());
+    }
+
+    public Iterator<ChannelFuture> iterator() {
+        return futures.values().iterator();
+    }
+
+    public synchronized boolean isDone() {
+        return done;
+    }
+
+    public synchronized boolean isCompleteSuccess() {
+        return done && successCount >= futures.size();
+    }
+
+    public synchronized boolean isPartialSuccess() {
+        return successCount > 0;
+    }
+
+    public synchronized boolean isPartialFailure() {
+        return failureCount > 0;
+    }
+
+    public synchronized boolean isCompleteFailure() {
+        return done && failureCount >= futures.size();
+    }
+
+    public void addListener(ChannelGroupFutureListener listener) {
+        if (listener == null) {
+            throw new NullPointerException("listener");
+        }
+
+        boolean notifyNow = false;
+        synchronized (this) {
+            if (done) {
+                notifyNow = true;
+            } else {
+                if (firstListener == null) {
+                    firstListener = listener;
+                } else {
+                    if (otherListeners == null) {
+                        otherListeners = new ArrayList<ChannelGroupFutureListener>(1);
+                    }
+                    otherListeners.add(listener);
+                }
+            }
+        }
+
+        if (notifyNow) {
+            notifyListener(listener);
+        }
+    }
+
+    public void removeListener(ChannelGroupFutureListener listener) {
+        if (listener == null) {
+            throw new NullPointerException("listener");
+        }
+
+        synchronized (this) {
+            if (!done) {
+                if (listener == firstListener) {
+                    if (otherListeners != null && !otherListeners.isEmpty()) {
+                        firstListener = otherListeners.remove(0);
+                    } else {
+                        firstListener = null;
+                    }
+                } else if (otherListeners != null) {
+                    otherListeners.remove(listener);
+                }
+            }
+        }
+    }
+
+    public ChannelGroupFuture await() throws InterruptedException {
+        synchronized (this) {
+            while (!done) {
+                waiters++;
+                try {
+                    this.wait();
+                } finally {
+                    waiters--;
+                }
+            }
+        }
+        return this;
+    }
+
+    public boolean await(long timeout, TimeUnit unit)
+            throws InterruptedException {
+        return await0(unit.toNanos(timeout), true);
+    }
+
+    public boolean await(long timeoutMillis) throws InterruptedException {
+        return await0(MILLISECONDS.toNanos(timeoutMillis), true);
+    }
+
+    public ChannelGroupFuture awaitUninterruptibly() {
+        synchronized (this) {
+            while (!done) {
+                waiters++;
+                try {
+                    this.wait();
+                } catch (InterruptedException e) {
+                    // Ignore.
+                } finally {
+                    waiters--;
+                }
+            }
+        }
+
+        return this;
+    }
+
+    public boolean awaitUninterruptibly(long timeout, TimeUnit unit) {
+        try {
+            return await0(unit.toNanos(timeout), false);
+        } catch (InterruptedException e) {
+            throw new InternalError();
+        }
+    }
+
+    public boolean awaitUninterruptibly(long timeoutMillis) {
+        try {
+            return await0(MILLISECONDS.toNanos(timeoutMillis), false);
+        } catch (InterruptedException e) {
+            throw new InternalError();
+        }
+    }
+
+    private boolean await0(long timeoutNanos, boolean interruptable) throws InterruptedException {
+        long startTime = timeoutNanos <= 0 ? 0 : System.nanoTime();
+        long waitTime = timeoutNanos;
+
+        synchronized (this) {
+            if (done) {
+                return done;
+            } else if (waitTime <= 0) {
+                return done;
+            }
+
+            waiters++;
+            try {
+                for (;;) {
+                    try {
+                        this.wait(waitTime / 1000000, (int) (waitTime % 1000000));
+                    } catch (InterruptedException e) {
+                        if (interruptable) {
+                            throw e;
+                        }
+                    }
+
+                    if (done) {
+                        return true;
+                    } else {
+                        waitTime = timeoutNanos - (System.nanoTime() - startTime);
+                        if (waitTime <= 0) {
+                            return done;
+                        }
+                    }
+                }
+            } finally {
+                waiters--;
+            }
+        }
+    }
+
+    boolean setDone() {
+        synchronized (this) {
+            // Allow only once.
+            if (done) {
+                return false;
+            }
+
+            done = true;
+            if (waiters > 0) {
+                notifyAll();
+            }
+        }
+
+        notifyListeners();
+        return true;
+    }
+
+    private void notifyListeners() {
+        // There won't be any visibility problem or concurrent modification
+        // because 'ready' flag will be checked against both addListener and
+        // removeListener calls.
+        if (firstListener != null) {
+            notifyListener(firstListener);
+            firstListener = null;
+
+            if (otherListeners != null) {
+                for (ChannelGroupFutureListener l: otherListeners) {
+                    notifyListener(l);
+                }
+                otherListeners = null;
+            }
+        }
+    }
+
+    private void notifyListener(ChannelGroupFutureListener l) {
+        try {
+            l.operationComplete(this);
+        } catch (Throwable t) {
+            logger.warn(
+                    "An exception was thrown by " +
+                    ChannelFutureListener.class.getSimpleName() + ".", t);
+        }
+    }
+}