Improvements for ContentCodec exceptions (#1428)

Motivation:

Align the order or arguments with standard JDK classes and throw these
types in all cases.

Modification:

- Swap `message` and `cause` arguments order to make them similar to the
superclass;
- Throw `CodecDecodingException/CodecEncodingException` when the codec is
closed;
- Remove unnecessary `requireNonNull` for aggregated encode and decode;

Result:

Aligned exceptions API and consistent exception type thrown from the codec.

servicetalk-encoding-api/src/main/java/io/servicetalk/encoding/api/CodecDecodingException.java

@@ -39,16 +39,17 @@ public CodecDecodingException(final ContentCodec codec, final String message) {
      * New instance.
      *
      * @param codec the codec in use.
-     * @param cause the cause of the exception.
      * @param message the reason of this exception.
+     * @param cause the cause of the exception.
      */
-    public CodecDecodingException(final ContentCodec codec, final Throwable cause, final String message) {
+    public CodecDecodingException(final ContentCodec codec, final String message, final Throwable cause) {
         super(message, cause);
         this.codec = codec;
     }
 
     /**
      * Returns the codec in use when this exception occurred.
+     *
      * @return the codec in use when this exception occurred.
      */
     public ContentCodec codec() {

servicetalk-encoding-api/src/main/java/io/servicetalk/encoding/api/CodecEncodingException.java

@@ -41,16 +41,17 @@ public CodecEncodingException(final ContentCodec codec, final String message) {
      * New instance.
      *
      * @param codec the codec in use.
-     * @param cause the cause of the exception.
      * @param message the reason of this exception.
+     * @param cause the cause of the exception.
      */
-    public CodecEncodingException(final ContentCodec codec, final Throwable cause, final String message) {
+    public CodecEncodingException(final ContentCodec codec, final String message, final Throwable cause) {
         super(message, cause);
         this.codec = requireNonNull(codec);
     }
 
     /**
      * Returns the codec in use when this exception occurred.
+     *
      * @return the codec in use when this exception occurred.
      */
     public ContentCodec codec() {

servicetalk-encoding-netty/src/main/java/io/servicetalk/encoding/netty/NettyChannelContentCodec.java

@@ -22,6 +22,7 @@
 import io.servicetalk.concurrent.api.Publisher;
 import io.servicetalk.encoding.api.CodecDecodingException;
 import io.servicetalk.encoding.api.CodecEncodingException;
+import io.servicetalk.encoding.api.ContentCodec;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler;
@@ -63,7 +64,6 @@ final class NettyChannelContentCodec extends AbstractContentCodec {
 
     @Override
     public Buffer encode(final Buffer src, final int offset, final int length, final BufferAllocator allocator) {
-        requireNonNull(src);
         requireNonNull(allocator);
 
         final Buffer slice = src.slice(src.readerIndex() + offset, length);
@@ -93,7 +93,7 @@ public Buffer encode(final Buffer src, final int offset, final int length, final
         } catch (CodecEncodingException e) {
             throw e;
         } catch (Throwable e) {
-            throw new CodecEncodingException(this, e, "");
+            throw wrapEncodingException(this, e);
         } finally {
             safeCleanup(channel);
         }
@@ -153,7 +153,7 @@ public void onNext(Buffer next) {
                                 subscription.request(1);
                             }
                         } catch (Throwable t) {
-                            throw new CodecEncodingException(NettyChannelContentCodec.this, t, "");
+                            throw wrapEncodingException(NettyChannelContentCodec.this, t);
                         }
                     }
 
@@ -168,7 +168,7 @@ public void onComplete() {
                         try {
                             cleanup(channel);
                         } catch (Throwable t) {
-                            subscriber.onError(new CodecEncodingException(NettyChannelContentCodec.this, t, ""));
+                            subscriber.onError(wrapEncodingException(NettyChannelContentCodec.this, t));
                             return;
                         }
 
@@ -179,7 +179,6 @@ public void onComplete() {
 
     @Override
     public Buffer decode(final Buffer src, final int offset, final int length, final BufferAllocator allocator) {
-        requireNonNull(src);
         requireNonNull(allocator);
 
         final Buffer slice = src.slice(src.readerIndex() + offset, length);
@@ -207,7 +206,7 @@ public Buffer decode(final Buffer src, final int offset, final int length, final
         } catch (CodecDecodingException e) {
             throw e;
         } catch (Throwable e) {
-            throw new CodecDecodingException(this, e, "");
+            throw wrapDecodingException(this, e);
         } finally {
             safeCleanup(channel);
         }
@@ -235,7 +234,8 @@ public void onSubscribe(final PublisherSource.Subscription subscription) {
             public void onNext(@Nullable final Buffer src) {
                 assert subscription != null;
                 if (!channel.isOpen()) {
-                    throw new IllegalStateException("Stream decoder previously closed but more input arrived");
+                    throw new CodecDecodingException(NettyChannelContentCodec.this,
+                            "Stream decoder previously closed but more input arrived");
                 }
 
                 if (src == null) {
@@ -254,7 +254,7 @@ public void onNext(@Nullable final Buffer src) {
                         subscription.request(1);
                     }
                 } catch (Throwable e) {
-                    throw new CodecDecodingException(NettyChannelContentCodec.this, e, "");
+                    throw wrapDecodingException(NettyChannelContentCodec.this, e);
                 }
             }
 
@@ -269,7 +269,7 @@ public void onComplete() {
                 try {
                     cleanup(channel);
                 } catch (Throwable t) {
-                    subscriber.onError(new CodecDecodingException(NettyChannelContentCodec.this, t, ""));
+                    subscriber.onError(wrapDecodingException(NettyChannelContentCodec.this, t));
                     return;
                 }
 
@@ -348,4 +348,12 @@ private static void safeCleanup(final EmbeddedChannel channel) {
             LOGGER.error("Error while closing embedded channel", t);
         }
     }
+
+    private static CodecEncodingException wrapEncodingException(final ContentCodec codec, final Throwable cause) {
+        return new CodecEncodingException(codec, "Unexpected exception during encoding", cause);
+    }
+
+    private static CodecDecodingException wrapDecodingException(final ContentCodec codec, final Throwable cause) {
+        return new CodecDecodingException(codec, "Unexpected exception during decoding", cause);
+    }
 }