Make sure ChannelFutureAggregator is thread-safe and only hold the lock

as short as possible. This also change it to lazy init the HashSet that
holds the ChannelFuture's. See #90

src/main/java/org/jboss/netty/channel/ChannelFutureAggregator.java

@@ -33,20 +33,24 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
 
     private final ChannelFuture aggregateFuture;
 
-    private final Set<ChannelFuture> pendingFutures;
+    private Set<ChannelFuture> pendingFutures;
 
     public ChannelFutureAggregator(ChannelFuture aggregateFuture) {
         this.aggregateFuture = aggregateFuture;
-        pendingFutures = new HashSet<ChannelFuture>();
     }
 
     public void addFuture(ChannelFuture future) {
+        synchronized(this) {
+            if (pendingFutures == null) {
+                pendingFutures = new HashSet<ChannelFuture>();
+            }
             pendingFutures.add(future);
+        }
         future.addListener(this);
     }
 
     @Override
-    public synchronized void operationComplete(ChannelFuture future)
+    public void operationComplete(ChannelFuture future)
             throws Exception {
         if (future.isCancelled()) {
             // TODO: what should the correct behaviour be when a fragment is cancelled?
@@ -54,17 +58,24 @@ public class ChannelFutureAggregator implements ChannelFutureListener {
             return;
         }
 
+        synchronized (this) {
+            if (pendingFutures == null) {
+                aggregateFuture.setSuccess();
+            } else {
                 pendingFutures.remove(future);
                 if (!future.isSuccess()) {
                     aggregateFuture.setFailure(future.getCause());
                     for (ChannelFuture pendingFuture: pendingFutures) {
                         pendingFuture.cancel();
                     }
-            return;
+                } else {
-        }
-
                     if (pendingFutures.isEmpty()) {
                         aggregateFuture.setSuccess();
                     }
                 }
             }
+
+        }
+        
+    }
+}