From 5526153459104bff2812c952a947576114419dd8 Mon Sep 17 00:00:00 2001 From: Norman Maurer Date: Fri, 4 Jan 2013 14:48:53 +0100 Subject: [PATCH] [#882] Add a PartialFlushException which will allow to notify the user that the flush/write was only partial succesful --- .../handler/codec/ByteToByteEncoder.java | 15 +++++++-- .../codec/MessageToMessageEncoder.java | 15 +++++++-- .../netty/channel/ChannelOutboundInvoker.java | 28 +++++++++++++++- .../netty/channel/PartialFlushException.java | 33 +++++++++++++++++++ 4 files changed, 84 insertions(+), 7 deletions(-) create mode 100644 transport/src/main/java/io/netty/channel/PartialFlushException.java diff --git a/codec/src/main/java/io/netty/handler/codec/ByteToByteEncoder.java b/codec/src/main/java/io/netty/handler/codec/ByteToByteEncoder.java index 5055e5d9a3..6d699be283 100644 --- a/codec/src/main/java/io/netty/handler/codec/ByteToByteEncoder.java +++ b/codec/src/main/java/io/netty/handler/codec/ByteToByteEncoder.java @@ -19,6 +19,7 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundByteHandlerAdapter; import io.netty.channel.ChannelPromise; +import io.netty.channel.PartialFlushException; /** * {@link ChannelOutboundByteHandlerAdapter} which encodes bytes in a stream-like fashion from one {@link ByteBuf} to an @@ -49,23 +50,31 @@ public abstract class ByteToByteEncoder extends ChannelOutboundByteHandlerAdapte public void flush(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { ByteBuf in = ctx.outboundByteBuffer(); ByteBuf out = ctx.nextOutboundByteBuffer(); + boolean encoded = false; while (in.readable()) { int oldInSize = in.readableBytes(); try { encode(ctx, in, out); + encoded = true; } catch (Throwable t) { + Throwable cause; if (t instanceof CodecException) { - ctx.fireExceptionCaught(t); + cause = t; } else { - ctx.fireExceptionCaught(new EncoderException(t)); + cause = new EncoderException(t); } + if (encoded) { + cause = new PartialFlushException("Unable to encoded all bytes", cause); + } + in.discardSomeReadBytes(); + promise.setFailure(cause); + return; } if (oldInSize == in.readableBytes()) { break; } } - in.discardSomeReadBytes(); ctx.flush(promise); } diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java index 21a2974f45..12ae8e6c45 100644 --- a/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java +++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageEncoder.java @@ -22,6 +22,7 @@ import io.netty.channel.ChannelOutboundMessageHandlerAdapter; import io.netty.channel.ChannelHandlerUtil; import io.netty.channel.ChannelPipeline; import io.netty.channel.ChannelPromise; +import io.netty.channel.PartialFlushException; /** * {@link ChannelOutboundMessageHandlerAdapter} which encodes from one message to an other message @@ -59,6 +60,8 @@ public abstract class MessageToMessageEncoder extends ChannelOutboundMessa @Override public void flush(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception { MessageBuf in = ctx.outboundMessageBuffer(); + boolean encoded = false; + for (;;) { try { Object msg = in.poll(); @@ -84,6 +87,7 @@ public abstract class MessageToMessageEncoder extends ChannelOutboundMessa if (omsg == imsg) { free = false; } + encoded = true; ChannelHandlerUtil.unfoldAndAdd(ctx, omsg, false); } finally { if (free) { @@ -91,14 +95,19 @@ public abstract class MessageToMessageEncoder extends ChannelOutboundMessa } } } catch (Throwable t) { + Throwable cause; if (t instanceof CodecException) { - ctx.fireExceptionCaught(t); + cause = t; } else { - ctx.fireExceptionCaught(new EncoderException(t)); + cause = new EncoderException(t); } + if (encoded) { + cause = new PartialFlushException("Unable to encoded all messages", cause); + } + promise.setFailure(cause); + return; } } - ctx.flush(promise); } diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java b/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java index bedaa17d3e..51fef953ab 100644 --- a/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java +++ b/transport/src/main/java/io/netty/channel/ChannelOutboundInvoker.java @@ -69,6 +69,13 @@ public interface ChannelOutboundInvoker { /** * Flush all pending data which belongs to this ChannelOutboundInvoker and notify the {@link ChannelFuture} * once the operation completes, either because the operation was successful or because of an error. + *

+ * Be aware that the flush could be only partially successful. In such cases the {@link ChannelFuture} will be + * failed with an {@link PartialFlushException}. So if you are interested to know if it was partial successful you + * need to check if the returned {@link ChannelFuture#cause()} returns an instance of + * {@link PartialFlushException}. In such cases you may want to call {@link #flush(ChannelPromise)} or + * {@link #flush()} to flush the rest of the data or just close the connection via {@link #close(ChannelPromise)} or + * {@link #close()} if it is not possible to recover. */ ChannelFuture flush(); @@ -76,7 +83,13 @@ public interface ChannelOutboundInvoker { * Write a message via this ChannelOutboundInvoker and notify the {@link ChannelFuture} * once the operation completes, either because the operation was successful or because of an error. * - * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)} + * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)}. + *

+ * Be aware that the write could be only partially successful as the message may need to get encoded before write it + * to the remote peer. In such cases the {@link ChannelFuture} will be failed with a {@link PartialFlushException}. + * In such cases you may want to call {@link #flush(ChannelPromise)} or {@link #flush()} to flush the rest of the + * data or just close the connection via {@link #close(ChannelPromise)} or {@link #close()} if it is not possible + * to recover. */ ChannelFuture write(Object message); @@ -152,6 +165,13 @@ public interface ChannelOutboundInvoker { /** * Flush all pending data which belongs to this ChannelOutboundInvoker and notify the {@link ChannelPromise} * once the operation completes, either because the operation was successful or because of an error. + *

+ * Be aware that the flush could be only partially successful. In such cases the {@link ChannelFuture} will be + * failed with an {@link PartialFlushException}. So if you are interested to know if it was partial successful you + * need to check if the returned {@link ChannelFuture#cause()} returns an instance of + * {@link PartialFlushException}. In such cases you may want to call {@link #flush(ChannelPromise)} or + * {@link #flush()} to flush the rest of the data or just close the connection via {@link #close(ChannelPromise)} or + * {@link #close()} if it is not possible to recover. * * The given {@link ChannelPromise} will be notified. */ @@ -163,6 +183,12 @@ public interface ChannelOutboundInvoker { * * If you want to write a {@link FileRegion} use {@link #sendFile(FileRegion)} * The given {@link ChannelPromise} will be notified and also returned. + *

+ * Be aware that the write could be only partially successful as the message may need to get encoded before write it + * to the remote peer. In such cases the {@link ChannelFuture} will be failed with a {@link PartialFlushException}. + * In such cases you may want to call {@link #flush(ChannelPromise)} or {@link #flush()} to flush the rest of the + * data or just close the connection via {@link #close(ChannelPromise)} or {@link #close()} if it is not possible + * to recover. */ ChannelFuture write(Object message, ChannelPromise promise); diff --git a/transport/src/main/java/io/netty/channel/PartialFlushException.java b/transport/src/main/java/io/netty/channel/PartialFlushException.java new file mode 100644 index 0000000000..f70e16e75e --- /dev/null +++ b/transport/src/main/java/io/netty/channel/PartialFlushException.java @@ -0,0 +1,33 @@ +/* + * Copyright 2013 The Netty Project + * + * The Netty Project 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 io.netty.channel; + +/** + * Special {@link RuntimeException} which will be used by {@link ChannelOutboundInvoker#flush(ChannelPromise)}, + * {@link ChannelOutboundInvoker#flush()}, {@link ChannelOutboundInvoker#write(Object)} and + * {@link ChannelOutboundInvoker#write(Object, ChannelPromise)} if the operation was only partial successful. + */ +public class PartialFlushException extends RuntimeException { + private static final long serialVersionUID = 990261865971015004L; + + public PartialFlushException(String msg, Throwable cause) { + super(msg, cause); + } + + public PartialFlushException(Throwable cause) { + super(cause); + } +}