Repository: netty/netty
Issue: Cannot use SCTP unordered flag
Labels: feature

Some applications, e.g. SIP over SCTP ([RFC4168](https://tools.ietf.org/html/rfc4168#section-5.1)), require the use of SCTP's unordered delivery flag.

Currently in Netty there is no way to set the unordered flag on an outgoing message. I assume the place to do this would be in `io.netty.channel.sctp.SctpMessage`, alongside the payload protocol ID and stream ID.

I'm happy to submit a PR but just wanted to check first in case there was some reason why this wasn't implemented. This applies to Netty 4.0 and later.


Issue comments:
@benevans Current io.netty.channel.sctp.SctpMessage is hiding some controls and, it could have been done better. Initially  it was designed to  work with SIGTRAN protocols and I didn't include ordering/completeness flags, which were not used at that time. Please feel free to send your pull request.

@benevans @jestan just be sure we need to keep API compat if you want to modify it for 4.0

@jestan I assigned it to you ;)

@normanmaurer sure, we will do it without breaking the API.

@jestan @benevans thanks! Looking forward to a PR !


Pull request review comments:

Patch that resolved the issue:
diff --git a/transport-sctp/src/main/java/io/netty/channel/sctp/SctpMessage.java b/transport-sctp/src/main/java/io/netty/channel/sctp/SctpMessage.java
index 6938f3bb3dd..f10f8eea56c 100644
--- a/transport-sctp/src/main/java/io/netty/channel/sctp/SctpMessage.java
+++ b/transport-sctp/src/main/java/io/netty/channel/sctp/SctpMessage.java
@@ -26,6 +26,7 @@
 public final class SctpMessage extends DefaultByteBufHolder {
     private final int streamIdentifier;
     private final int protocolIdentifier;
+    private final boolean unordered;
 
     private final MessageInfo msgInfo;
 
@@ -36,9 +37,21 @@ public final class SctpMessage extends DefaultByteBufHolder {
      * @param payloadBuffer channel buffer
      */
     public SctpMessage(int protocolIdentifier, int streamIdentifier, ByteBuf payloadBuffer) {
+        this(protocolIdentifier, streamIdentifier, false, payloadBuffer);
+    }
+
+    /**
+     * Essential data that is being carried within SCTP Data Chunk
+     * @param protocolIdentifier of payload
+     * @param streamIdentifier that you want to send the payload
+     * @param unordered if {@literal true}, the SCTP Data Chunk will be sent with the U (unordered) flag set.
+     * @param payloadBuffer channel buffer
+     */
+    public SctpMessage(int protocolIdentifier, int streamIdentifier, boolean unordered, ByteBuf payloadBuffer) {
         super(payloadBuffer);
         this.protocolIdentifier = protocolIdentifier;
         this.streamIdentifier = streamIdentifier;
+        this.unordered = unordered;
         msgInfo = null;
     }
 
@@ -55,6 +68,7 @@ public SctpMessage(MessageInfo msgInfo, ByteBuf payloadBuffer) {
         this.msgInfo = msgInfo;
         streamIdentifier = msgInfo.streamNumber();
         protocolIdentifier = msgInfo.payloadProtocolID();
+        unordered = msgInfo.isUnordered();
     }
 
     /**
@@ -71,6 +85,13 @@ public int protocolIdentifier() {
         return protocolIdentifier;
     }
 
+    /**
+     * return the unordered flag
+     */
+    public boolean isUnordered() {
+        return unordered;
+    }
+
     /**
      * Return the {@link MessageInfo} for inbound messages or {@code null} for
      * outbound messages.
@@ -111,6 +132,10 @@ public boolean equals(Object o) {
             return false;
         }
 
+        if (unordered != sctpFrame.unordered) {
+            return false;
+        }
+
         if (!content().equals(sctpFrame.content())) {
             return false;
         }
@@ -129,7 +154,7 @@ public int hashCode() {
     @Override
     public SctpMessage copy() {
         if (msgInfo == null) {
-            return new SctpMessage(protocolIdentifier, streamIdentifier, content().copy());
+            return new SctpMessage(protocolIdentifier, streamIdentifier, unordered, content().copy());
         } else {
             return new SctpMessage(msgInfo, content().copy());
         }
@@ -138,7 +163,7 @@ public SctpMessage copy() {
     @Override
     public SctpMessage duplicate() {
         if (msgInfo == null) {
-            return new SctpMessage(protocolIdentifier, streamIdentifier, content().duplicate());
+            return new SctpMessage(protocolIdentifier, streamIdentifier, unordered, content().duplicate());
         } else {
             return new SctpMessage(msgInfo, content().copy());
         }
@@ -161,10 +186,12 @@ public String toString() {
         if (refCnt() == 0) {
             return "SctpFrame{" +
                     "streamIdentifier=" + streamIdentifier + ", protocolIdentifier=" + protocolIdentifier +
+                    ", unordered=" + unordered +
                     ", data=(FREED)}";
         }
         return "SctpFrame{" +
                 "streamIdentifier=" + streamIdentifier + ", protocolIdentifier=" + protocolIdentifier +
+                ", unordered=" + unordered +
                 ", data=" + ByteBufUtil.hexDump(content()) + '}';
     }
 }
diff --git a/transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java b/transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java
index 8542ce5fbef..8ce914a2b65 100644
--- a/transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java
+++ b/transport-sctp/src/main/java/io/netty/channel/sctp/nio/NioSctpChannel.java
@@ -320,6 +320,7 @@ protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Ex
         final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
         mi.payloadProtocolID(packet.protocolIdentifier());
         mi.streamNumber(packet.streamIdentifier());
+        mi.unordered(packet.isUnordered());
 
         final int writtenBytes = javaChannel().send(nioData, mi);
         return writtenBytes > 0;
@@ -334,7 +335,8 @@ protected final Object filterOutboundMessage(Object msg) throws Exception {
                 return m;
             }
 
-            return new SctpMessage(m.protocolIdentifier(), m.streamIdentifier(), newDirectBuffer(m, buf));
+            return new SctpMessage(m.protocolIdentifier(), m.streamIdentifier(), m.isUnordered(),
+                                   newDirectBuffer(m, buf));
         }
 
         throw new UnsupportedOperationException(
diff --git a/transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java b/transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java
index e12981f7a6e..6416ea4e60e 100755
--- a/transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java
+++ b/transport-sctp/src/main/java/io/netty/channel/sctp/oio/OioSctpChannel.java
@@ -263,6 +263,7 @@ protected void doWrite(ChannelOutboundBuffer in) throws Exception {
                 final MessageInfo mi = MessageInfo.createOutgoing(association(), null, packet.streamIdentifier());
                 mi.payloadProtocolID(packet.protocolIdentifier());
                 mi.streamNumber(packet.streamIdentifier());
+                mi.unordered(packet.isUnordered());
 
                 ch.send(nioData, mi);
                 written ++;
diff --git a/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpMessageCompletionHandler.java b/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpMessageCompletionHandler.java
index 30559e501f2..01f5ae3c4f8 100644
--- a/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpMessageCompletionHandler.java
+++ b/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpMessageCompletionHandler.java
@@ -41,6 +41,7 @@ protected void decode(ChannelHandlerContext ctx, SctpMessage msg, List<Object> o
         final int protocolIdentifier = msg.protocolIdentifier();
         final int streamIdentifier = msg.streamIdentifier();
         final boolean isComplete = msg.isComplete();
+        final boolean isUnordered = msg.isUnordered();
 
         ByteBuf frag;
         if (fragments.containsKey(streamIdentifier)) {
@@ -61,6 +62,7 @@ protected void decode(ChannelHandlerContext ctx, SctpMessage msg, List<Object> o
             SctpMessage assembledMsg = new SctpMessage(
                     protocolIdentifier,
                     streamIdentifier,
+                    isUnordered,
                     Unpooled.wrappedBuffer(frag, byteBuf));
             out.add(assembledMsg);
         } else {
diff --git a/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpOutboundByteStreamHandler.java b/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpOutboundByteStreamHandler.java
index 26cecaca463..1cd098be58c 100644
--- a/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpOutboundByteStreamHandler.java
+++ b/transport-sctp/src/main/java/io/netty/handler/codec/sctp/SctpOutboundByteStreamHandler.java
@@ -25,23 +25,34 @@
 /**
  * A ChannelHandler which transform {@link ByteBuf} to {@link SctpMessage}  and send it through a specific stream
  * with given protocol identifier.
- *
+ * Unordered delivery of all messages may be requested by passing unordered = true to the constructor.
  */
 public class SctpOutboundByteStreamHandler extends MessageToMessageEncoder<ByteBuf> {
     private final int streamIdentifier;
     private final int protocolIdentifier;
+    private final boolean unordered;
 
     /**
      * @param streamIdentifier      stream number, this should be >=0 or <= max stream number of the association.
      * @param protocolIdentifier    supported application protocol id.
      */
     public SctpOutboundByteStreamHandler(int streamIdentifier, int protocolIdentifier) {
+        this(streamIdentifier, protocolIdentifier, false);
+    }
+
+    /**
+     * @param streamIdentifier      stream number, this should be >=0 or <= max stream number of the association.
+     * @param protocolIdentifier    supported application protocol id.
+     * @param unordered             if {@literal true}, SCTP Data Chunks will be sent with the U (unordered) flag set.
+     */
+    public SctpOutboundByteStreamHandler(int streamIdentifier, int protocolIdentifier, boolean unordered) {
         this.streamIdentifier = streamIdentifier;
         this.protocolIdentifier = protocolIdentifier;
+        this.unordered = unordered;
     }
 
     @Override
     protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
-        out.add(new SctpMessage(streamIdentifier, protocolIdentifier, msg.retain()));
+        out.add(new SctpMessage(streamIdentifier, protocolIdentifier, unordered, msg.retain()));
     }
 }


Test patch:
diff --git a/testsuite/src/main/java/io/netty/testsuite/transport/sctp/SctpEchoTest.java b/testsuite/src/main/java/io/netty/testsuite/transport/sctp/SctpEchoTest.java
index 0b126b8a136..067d8e6a671 100644
--- a/testsuite/src/main/java/io/netty/testsuite/transport/sctp/SctpEchoTest.java
+++ b/testsuite/src/main/java/io/netty/testsuite/transport/sctp/SctpEchoTest.java
@@ -53,10 +53,20 @@ public void testSimpleEcho() throws Throwable {
     }
 
     public void testSimpleEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
-        testSimpleEcho0(sb, cb);
+        testSimpleEcho0(sb, cb, false);
     }
 
-    private static void testSimpleEcho0(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+    @Test
+    public void testSimpleEchoUnordered() throws Throwable {
+        Assume.assumeTrue(TestUtils.isSctpSupported());
+        run();
+    }
+
+    public void testSimpleEchoUnordered(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        testSimpleEcho0(sb, cb, true);
+    }
+
+    private static void testSimpleEcho0(ServerBootstrap sb, Bootstrap cb, final boolean unordered) throws Throwable {
         final EchoHandler sh = new EchoHandler();
         final EchoHandler ch = new EchoHandler();
 
@@ -66,7 +76,7 @@ public void initChannel(SctpChannel c) throws Exception {
                 c.pipeline().addLast(
                         new SctpMessageCompletionHandler(),
                         new SctpInboundByteStreamHandler(0, 0),
-                        new SctpOutboundByteStreamHandler(0, 0),
+                        new SctpOutboundByteStreamHandler(0, 0, unordered),
                         sh);
             }
         });
@@ -76,7 +86,7 @@ public void initChannel(SctpChannel c) throws Exception {
                 c.pipeline().addLast(
                         new SctpMessageCompletionHandler(),
                         new SctpInboundByteStreamHandler(0, 0),
-                        new SctpOutboundByteStreamHandler(0, 0),
+                        new SctpOutboundByteStreamHandler(0, 0, unordered),
                         ch);
             }
         });