Adjust ChannelGroup to behave the same like a Channel in terms of write and flush

This commit is contained in:
Norman Maurer 2013-07-10 08:27:11 +02:00
parent da5c6add14
commit dd763698dc
2 changed files with 40 additions and 4 deletions

View File

@ -103,10 +103,24 @@ public interface ChannelGroup extends Set<Channel>, Comparable<ChannelGroup> {
* condition. Please note that this operation is asynchronous as
* {@link Channel#write(Object)} is.
*
* @return itself
*/
ChannelGroup write(Object message);
/**
* Flush all {@link Channel}s in this
* group. Please note that this operation is asynchronous as
* {@link Channel#flush()} is.
*
* @return the {@link ChannelGroupFuture} instance that notifies when
* the operation is done for all channels
*/
ChannelGroupFuture write(Object message);
ChannelGroupFuture flush();
/**
* Shortcut for calling {@link #write(Object)} and {@link #flush()}.
*/
ChannelGroupFuture flushAndWrite(Object message);
/**
* Disconnects all {@link Channel}s in this group from their remote peers.

View File

@ -192,17 +192,39 @@ public class DefaultChannelGroup extends AbstractSet<Channel> implements Channel
}
@Override
public ChannelGroupFuture write(Object message) {
public ChannelGroup write(Object message) {
if (message == null) {
throw new NullPointerException("message");
}
Map<Channel, ChannelFuture> futures = new LinkedHashMap<Channel, ChannelFuture>(size());
for (Channel c: nonServerChannels) {
futures.put(c, c.write(safeDuplicate(message)).flush());
c.write(safeDuplicate(message));
}
ReferenceCountUtil.release(message);
return this;
}
@Override
public ChannelGroupFuture flush() {
Map<Channel, ChannelFuture> futures = new LinkedHashMap<Channel, ChannelFuture>(size());
for (Channel c: nonServerChannels) {
futures.put(c, c.flush());
}
return new DefaultChannelGroupFuture(this, futures, executor);
}
@Override
public ChannelGroupFuture flushAndWrite(Object message) {
Map<Channel, ChannelFuture> futures = new LinkedHashMap<Channel, ChannelFuture>(size());
for (Channel c: nonServerChannels) {
futures.put(c, c.writeAndFlush(safeDuplicate(message)));
}
ReferenceCountUtil.release(message);
return new DefaultChannelGroupFuture(this, futures, executor);
}