Merge pull request #633 from jestan/master

SCTP Transport Codecs and TestCases

codec/src/main/java/io/netty/handler/codec/sctp/SctpInboundByteStreamHandler.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2012 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.handler.codec.sctp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundMessageHandlerAdapter;
+import io.netty.channel.socket.SctpMessage;
+import io.netty.handler.codec.CodecException;
+
+/**
+ * A ChannelHandler which receives  {@link SctpMessage} belongs to a application protocol form a specific SCTP Stream
+ * and decode it as {@link ByteBuf}.
+ */
+public class SctpInboundByteStreamHandler extends ChannelInboundMessageHandlerAdapter<SctpMessage> {
+    private final int protocolIdentifier;
+    private final int streamIdentifier;
+
+
+    /**
+     * @param streamIdentifier   accepted stream number, this should be >=0 or <= max stream number of the association.
+     * @param protocolIdentifier supported application protocol.
+     */
+    public SctpInboundByteStreamHandler(int protocolIdentifier, int streamIdentifier) {
+        this.protocolIdentifier = protocolIdentifier;
+        this.streamIdentifier = streamIdentifier;
+    }
+
+    protected boolean isDecodable(SctpMessage msg) {
+        return msg.getProtocolIdentifier() == protocolIdentifier && msg.getStreamIdentifier() == streamIdentifier;
+    }
+
+    @Override
+    public void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
+        if (!isDecodable(msg)) {
+            ctx.nextInboundMessageBuffer().add(msg);
+            ctx.fireInboundBufferUpdated();
+            return;
+        }
+
+        if (!msg.isComplete()) {
+            throw new CodecException(String.format("Received SctpMessage is not complete, please add %s in the " +
+                    "pipeline before this handler", SctpMessageCompletionHandler.class.getSimpleName()));
+        }
+
+        ctx.nextInboundByteBuffer().writeBytes(msg.getPayloadBuffer());
+        ctx.fireInboundBufferUpdated();
+    }
+}

codec/src/main/java/io/netty/handler/codec/sctp/SctpMessageCompletionHandler.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2012 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.handler.codec.sctp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundMessageHandlerAdapter;
+import io.netty.channel.socket.SctpMessage;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class SctpMessageCompletionHandler extends ChannelInboundMessageHandlerAdapter<SctpMessage> {
+    private Map<Integer, ByteBuf> fragments = new HashMap<Integer, ByteBuf>();
+
+    /**
+     */
+    public SctpMessageCompletionHandler() {
+    }
+
+    @Override
+    public void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
+
+        final ByteBuf byteBuf = msg.getPayloadBuffer();
+        final int protocolIdentifier = msg.getProtocolIdentifier();
+        final int streamIdentifier = msg.getStreamIdentifier();
+        final boolean isComplete = msg.isComplete();
+
+        ByteBuf frag;
+
+        if (fragments.containsKey(streamIdentifier)) {
+            frag = fragments.remove(streamIdentifier);
+        } else {
+            frag = Unpooled.EMPTY_BUFFER;
+        }
+
+        if (isComplete && !frag.readable()) {
+            //data chunk is not fragmented
+            fireAssembledMessage(ctx, msg);
+        } else if (!isComplete && frag.readable()) {
+            //more message to complete
+            fragments.put(streamIdentifier, Unpooled.wrappedBuffer(frag, byteBuf));
+        } else if (isComplete && frag.readable()) {
+            //last message to complete
+            fragments.remove(streamIdentifier);
+            SctpMessage assembledMsg = new SctpMessage(
+                    protocolIdentifier,
+                    streamIdentifier,
+                    Unpooled.wrappedBuffer(frag, byteBuf));
+            fireAssembledMessage(ctx, assembledMsg);
+        } else {
+            //first incomplete message
+            fragments.put(streamIdentifier, byteBuf);
+        }
+    }
+
+    protected void fireAssembledMessage(ChannelHandlerContext ctx, SctpMessage assembledMsg) {
+        ctx.nextInboundMessageBuffer().add(assembledMsg);
+        ctx.fireInboundBufferUpdated();
+    }
+}

codec/src/main/java/io/netty/handler/codec/sctp/SctpMessageToMessageDecoder.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2012 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.handler.codec.sctp;
+
+import io.netty.channel.socket.SctpMessage;
+import io.netty.handler.codec.CodecException;
+import io.netty.handler.codec.MessageToMessageDecoder;
+
+public abstract class SctpMessageToMessageDecoder<O> extends MessageToMessageDecoder<SctpMessage, O> {
+
+    @Override
+    public boolean isDecodable(Object msg) throws Exception {
+        if (msg instanceof SctpMessage) {
+            SctpMessage sctpMsg = (SctpMessage) msg;
+            if (sctpMsg.isComplete()) {
+                return true;
+            }
+
+            throw new CodecException(String.format("Received SctpMessage is not complete, please add %s in " +
+                    "the pipeline before this handler", SctpMessageCompletionHandler.class.getSimpleName()));
+        } else {
+            return false;
+        }
+    }
+}

codec/src/main/java/io/netty/handler/codec/sctp/SctpMessageToMessageEncoder.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2012 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.handler.codec.sctp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.socket.SctpMessage;
+import io.netty.handler.codec.MessageToMessageEncoder;
+
+public abstract class SctpMessageToMessageEncoder<I> extends MessageToMessageEncoder<I, SctpMessage> {
+
+    /**
+     * Returns {@code true} if and only if the specified message can be encoded by this encoder.
+     *
+     * @param msg the message
+     */
+    public boolean isEncodable(Object msg) throws Exception {
+        return true;
+    }
+
+    public abstract SctpMessage encode(ChannelHandlerContext ctx, I msg) throws Exception;
+}

codec/src/main/java/io/netty/handler/codec/sctp/SctpOutboundByteStreamHandler.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2012 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.handler.codec.sctp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.MessageBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundByteHandlerAdapter;
+import io.netty.channel.socket.SctpMessage;
+import io.netty.handler.codec.EncoderException;
+
+/**
+ * A ChannelHandler which transform {@link ByteBuf} to {@link SctpMessage}  and send it through a specific stream
+ * with given protocol identifier.
+ *
+ */
+public class SctpOutboundByteStreamHandler extends ChannelOutboundByteHandlerAdapter {
+    private final int streamIdentifier;
+    private final int protocolIdentifier;
+
+
+    /**
+     * @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 = streamIdentifier;
+        this.protocolIdentifier = protocolIdentifier;
+    }
+
+    @Override
+    public void flush(ChannelHandlerContext ctx, ChannelFuture future) throws Exception {
+        ByteBuf in = ctx.outboundByteBuffer();
+
+        try {
+            MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
+            ByteBuf payload = Unpooled.buffer(in.readableBytes());
+            payload.writeBytes(in);
+            out.add(new SctpMessage(streamIdentifier, protocolIdentifier, payload));
+            in.discardReadBytes();
+        } catch (Throwable t) {
+            ctx.fireExceptionCaught(new EncoderException(t));
+        }
+
+        ctx.flush(future);
+    }
+}

codec/src/main/java/io/netty/handler/codec/sctp/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012 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.
+ */
+
+/**
+ * Decoder and encoders to manage message completion and multi-streaming codec in SCTP/IP.
+ */
+package io.netty.handler.codec.sctp;

example/src/main/java/io/netty/example/sctp/NioSctpEchoClient.java

@@ -86,7 +86,7 @@ public class NioSctpEchoClient {
         }
 
         // Parse options.
-        final String host = "localhost";
+        final String host = args[0];
         final int port = Integer.parseInt(args[1]);
         final int firstMessageSize;
         if (args.length == 3) {

example/src/main/java/io/netty/example/sctp/OioSctpEchoClient.java

@@ -85,7 +85,7 @@ public class OioSctpEchoClient {
         }
 
         // Parse options.
-        final String host = "localhost";
+        final String host = args[0];
         final int port = Integer.parseInt(args[1]);
         final int firstMessageSize;
         if (args.length == 3) {

example/src/main/java/io/netty/example/sctp/SctpEchoClientHandler.java

@@ -20,7 +20,6 @@ import io.netty.buffer.MessageBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundMessageHandlerAdapter;
-import io.netty.channel.socket.SctpData;
 import io.netty.channel.socket.SctpMessage;
 
 import java.util.logging.Level;
@@ -53,16 +52,14 @@ public class SctpEchoClientHandler extends ChannelInboundMessageHandlerAdapter<S
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) {
-        ctx.write(new SctpData(0, 0, firstMessage));
+        ctx.write(new SctpMessage(0, 0, firstMessage));
     }
 
     @Override
     public void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
-        if (msg instanceof SctpData) {
+        MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
-            MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
+        out.add(msg);
-            out.add(msg);
+        ctx.flush();
-            ctx.flush();
-        }
     }
 
     @Override

example/src/main/java/io/netty/example/sctp/SctpEchoServerHandler.java

@@ -19,7 +19,6 @@ import io.netty.buffer.MessageBuf;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundMessageHandlerAdapter;
-import io.netty.channel.socket.SctpData;
 import io.netty.channel.socket.SctpMessage;
 
 import java.util.logging.Level;
@@ -43,10 +42,8 @@ public class SctpEchoServerHandler extends ChannelInboundMessageHandlerAdapter<S
 
     @Override
     public void messageReceived(ChannelHandlerContext ctx, SctpMessage msg) throws Exception {
-        if (msg instanceof SctpData) {
+        MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
-            MessageBuf<Object> out = ctx.nextOutboundMessageBuffer();
+        out.add(msg);
-            out.add(msg);
+        ctx.flush();
-            ctx.flush();
-        }
     }
 }

testsuite/src/test/java/io/netty/testsuite/transport/socket/AbstractSctpTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2012 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.testsuite.transport.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.logging.InternalLogger;
+import io.netty.logging.InternalLoggerFactory;
+import io.netty.testsuite.transport.socket.SocketTestPermutation.Factory;
+import io.netty.testsuite.util.TestUtils;
+import io.netty.util.NetworkConstants;
+import org.junit.Rule;
+import org.junit.rules.TestName;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.Map.Entry;
+
+public abstract class AbstractSctpTest {
+
+    private static final List<Entry<Factory<ServerBootstrap>, Factory<Bootstrap>>> COMBO =
+            SocketTestPermutation.sctpChannel();
+
+    @Rule
+    public final TestName testName = new TestName();
+
+    protected final InternalLogger logger = InternalLoggerFactory.getInstance(getClass());
+
+    protected volatile ServerBootstrap sb;
+    protected volatile Bootstrap cb;
+    protected volatile InetSocketAddress addr;
+    protected volatile Factory<Bootstrap> currentBootstrap;
+    
+    protected void run() throws Throwable {
+        int i = 0;
+        for (Entry<Factory<ServerBootstrap>, Factory<Bootstrap>> e: COMBO) {
+            currentBootstrap = e.getValue();
+            sb = e.getKey().newInstance();
+            cb = e.getValue().newInstance();
+            addr = new InetSocketAddress(
+                    NetworkConstants.LOCALHOST, TestUtils.getFreePort());
+            sb.localAddress(addr);
+            cb.remoteAddress(addr);
+
+            logger.info(String.format(
+                    "Running: %s %d of %d", testName.getMethodName(), ++ i, COMBO.size()));
+            try {
+                Method m = getClass().getDeclaredMethod(
+                        testName.getMethodName(), ServerBootstrap.class, Bootstrap.class);
+                m.invoke(this, sb, cb);
+            } catch (InvocationTargetException ex) {
+                throw ex.getCause();
+            } finally {
+                sb.shutdown();
+                cb.shutdown();
+            }
+        }
+    }
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/SctpEchoTest.java

@@ -0,0 +1,201 @@
+/*
+ * Copyright 2012 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.testsuite.transport.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundByteHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SctpChannel;
+import io.netty.handler.codec.sctp.SctpInboundByteStreamHandler;
+import io.netty.handler.codec.sctp.SctpMessageCompletionHandler;
+import io.netty.handler.codec.sctp.SctpOutboundByteStreamHandler;
+import io.netty.testsuite.util.TestUtils;
+import org.junit.Assume;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+
+public class SctpEchoTest extends AbstractSctpTest {
+
+    private static final Random random = new Random();
+    static final byte[] data = new byte[4096];//could not test ultra jumbo frames
+
+    static {
+        random.nextBytes(data);
+    }
+
+    @Test
+    public void testSimpleEcho() throws Throwable {
+        Assume.assumeTrue(TestUtils.isSctpSupported());
+        run();
+    }
+
+    public void testSimpleEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        testSimpleEcho0(sb, cb, Integer.MAX_VALUE);
+    }
+
+    @Test
+    @Ignore("TODO: fix this after OioSctp EventLoop done")
+    public void testSimpleEchoWithBoundedBuffer() throws Throwable {
+        Assume.assumeTrue(TestUtils.isSctpSupported());
+        run();
+    }
+
+    public void testSimpleEchoWithBoundedBuffer(ServerBootstrap sb, Bootstrap cb) throws Throwable {
+        testSimpleEcho0(sb, cb, 32);
+    }
+
+    private static void testSimpleEcho0(ServerBootstrap sb, Bootstrap cb, int maxInboundBufferSize) throws Throwable {
+        final EchoHandler sh = new EchoHandler(maxInboundBufferSize);
+        final EchoHandler ch = new EchoHandler(maxInboundBufferSize);
+
+        sb.childHandler(new ChannelInitializer<SctpChannel>() {
+            @Override
+            public void initChannel(SctpChannel c) throws Exception {
+                c.pipeline().addLast(
+                        new SctpMessageCompletionHandler(),
+                        new SctpInboundByteStreamHandler(0, 0),
+                        new SctpOutboundByteStreamHandler(0, 0),
+                        sh);
+            }
+        });
+        cb.handler(new ChannelInitializer<SctpChannel>() {
+            @Override
+            public void initChannel(SctpChannel c) throws Exception {
+                c.pipeline().addLast(
+                        new SctpMessageCompletionHandler(),
+                        new SctpInboundByteStreamHandler(0, 0),
+                        new SctpOutboundByteStreamHandler(0, 0),
+                        ch);
+            }
+        });
+
+        Channel sc = sb.bind().sync().channel();
+        Channel cc = cb.connect().sync().channel();
+
+        for (int i = 0; i < data.length; ) {
+            int length = Math.min(random.nextInt(1024 * 64), data.length - i);
+            cc.write(Unpooled.wrappedBuffer(data, i, length));
+            i += length;
+        }
+
+        while (ch.counter < data.length) {
+            if (sh.exception.get() != null) {
+                break;
+            }
+            if (ch.exception.get() != null) {
+                break;
+            }
+
+            try {
+                Thread.sleep(1);
+            } catch (InterruptedException e) {
+                // Ignore.
+            }
+        }
+
+        while (sh.counter < data.length) {
+            if (sh.exception.get() != null) {
+                break;
+            }
+            if (ch.exception.get() != null) {
+                break;
+            }
+
+            try {
+                Thread.sleep(1);
+            } catch (InterruptedException e) {
+                // Ignore.
+            }
+        }
+
+        sh.channel.close().sync();
+        ch.channel.close().sync();
+        sc.close().sync();
+
+        if (sh.exception.get() != null && !(sh.exception.get() instanceof IOException)) {
+            throw sh.exception.get();
+        }
+        if (ch.exception.get() != null && !(ch.exception.get() instanceof IOException)) {
+            throw ch.exception.get();
+        }
+        if (sh.exception.get() != null) {
+            throw sh.exception.get();
+        }
+        if (ch.exception.get() != null) {
+            throw ch.exception.get();
+        }
+    }
+
+    private static class EchoHandler extends ChannelInboundByteHandlerAdapter {
+        private final int maxInboundBufferSize;
+        volatile Channel channel;
+        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+        volatile int counter;
+
+        EchoHandler(int maxInboundBufferSize) {
+            this.maxInboundBufferSize = maxInboundBufferSize;
+        }
+
+        @Override
+        public ByteBuf newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
+            return Unpooled.buffer(0, maxInboundBufferSize);
+        }
+
+        @Override
+        public void channelActive(ChannelHandlerContext ctx)
+                throws Exception {
+            channel = ctx.channel();
+        }
+
+        @Override
+        public void inboundBufferUpdated(
+                ChannelHandlerContext ctx, ByteBuf in)
+                throws Exception {
+            byte[] actual = new byte[in.readableBytes()];
+            in.readBytes(actual);
+
+            int lastIdx = counter;
+            for (int i = 0; i < actual.length; i++) {
+                assertEquals(data[i + lastIdx], actual[i]);
+            }
+
+            if (channel.parent() != null) {
+                channel.write(Unpooled.wrappedBuffer(actual));
+            }
+
+            counter += actual.length;
+        }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx,
+                                    Throwable cause) throws Exception {
+            if (exception.compareAndSet(null, cause)) {
+                ctx.close();
+            }
+        }
+    }
+}

testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketTestPermutation.java

@@ -26,11 +26,15 @@ import io.netty.channel.socket.aio.AioSocketChannel;
 import io.netty.channel.socket.nio.NioDatagramChannel;
 import io.netty.channel.socket.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.channel.socket.nio.NioSctpServerChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.channel.socket.nio.NioSctpChannel;
 import io.netty.channel.socket.oio.OioDatagramChannel;
 import io.netty.channel.socket.oio.OioEventLoopGroup;
 import io.netty.channel.socket.oio.OioServerSocketChannel;
+import io.netty.channel.socket.oio.OioSctpServerChannel;
 import io.netty.channel.socket.oio.OioSocketChannel;
+import io.netty.channel.socket.oio.OioSctpChannel;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -130,6 +134,43 @@ final class SocketTestPermutation {
         return list;
     }
 
+    static List<Entry<Factory<ServerBootstrap>, Factory<Bootstrap>>> sctpChannel() {
+        List<Entry<Factory<ServerBootstrap>, Factory<Bootstrap>>> list =
+                new ArrayList<Entry<Factory<ServerBootstrap>, Factory<Bootstrap>>>();
+
+        // Make the list of SCTP ServerBootstrap factories.
+        List<Factory<ServerBootstrap>> sbfs = sctpServerChannel();
+
+        // Make the list of SCTP Bootstrap factories.
+        List<Factory<Bootstrap>> cbfs = sctpClientChannel();
+
+        // Populate the combinations
+        for (Factory<ServerBootstrap> sbf: sbfs) {
+            for (Factory<Bootstrap> cbf: cbfs) {
+                final Factory<ServerBootstrap> sbf0 = sbf;
+                final Factory<Bootstrap> cbf0 = cbf;
+                list.add(new Entry<Factory<ServerBootstrap>, Factory<Bootstrap>>() {
+                    @Override
+                    public Factory<ServerBootstrap> getKey() {
+                        return sbf0;
+                    }
+
+                    @Override
+                    public Factory<Bootstrap> getValue() {
+                        return cbf0;
+                    }
+
+                    @Override
+                    public Factory<Bootstrap> setValue(Factory<Bootstrap> value) {
+                        throw new UnsupportedOperationException();
+                    }
+                });
+            }
+        }
+
+        return list;
+    }
+
     static List<Factory<ServerBootstrap>> serverSocket() {
         List<Factory<ServerBootstrap>> list = new ArrayList<Factory<ServerBootstrap>>();
 
@@ -197,6 +238,47 @@ final class SocketTestPermutation {
         return list;
     }
 
+    static List<Factory<ServerBootstrap>> sctpServerChannel() {
+        List<Factory<ServerBootstrap>> list = new ArrayList<Factory<ServerBootstrap>>();
+
+        // Make the list of ServerBootstrap factories.
+        list.add(new Factory<ServerBootstrap>() {
+            @Override
+            public ServerBootstrap newInstance() {
+                return new ServerBootstrap().
+                        group(new NioEventLoopGroup(), new NioEventLoopGroup()).
+                        channel(NioSctpServerChannel.class);
+            }
+        });
+        list.add(new Factory<ServerBootstrap>() {
+            @Override
+            public ServerBootstrap newInstance() {
+                return new ServerBootstrap().
+                        group(new OioEventLoopGroup(), new OioEventLoopGroup()).
+                        channel(OioSctpServerChannel.class);
+            }
+        });
+
+        return list;
+    }
+
+    static List<Factory<Bootstrap>> sctpClientChannel() {
+        List<Factory<Bootstrap>> list = new ArrayList<Factory<Bootstrap>>();
+        list.add(new Factory<Bootstrap>() {
+            @Override
+            public Bootstrap newInstance() {
+                return new Bootstrap().group(new NioEventLoopGroup()).channel(NioSctpChannel.class);
+            }
+        });
+        list.add(new Factory<Bootstrap>() {
+            @Override
+            public Bootstrap newInstance() {
+                return new Bootstrap().group(new OioEventLoopGroup()).channel(OioSctpChannel.class);
+            }
+        });
+        return list;
+    }
+
     private SocketTestPermutation() {}
 
     interface Factory<T> {

testsuite/src/test/java/io/netty/testsuite/util/TestUtils.java

@@ -15,15 +15,13 @@
  */
 package io.netty.testsuite.util;
 
+import com.sun.nio.sctp.SctpChannel;
 import io.netty.util.NetworkConstants;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
 
 public class TestUtils {
 
@@ -78,5 +76,29 @@ public class TestUtils {
         throw new RuntimeException("unable to find a free port");
     }
 
+    /**
+     * Return <code>true</code> if SCTP is supported by the running os.
+     *
+     */
+    public static boolean isSctpSupported() {
+        String os = System.getProperty("os.name").toLowerCase(Locale.UK);
+        if (os.equals("unix") || os.equals("linux") || os.equals("sun") || os.equals("solaris")) {
+            try {
+                SctpChannel.open();
+            } catch (IOException e) {
+                // ignore
+            } catch (UnsupportedOperationException e) {
+                // This exception may get thrown if the OS does not have
+                // the shared libs installed.
+                System.out.print("Not supported: " + e.getMessage());
+                return false;
+
+            }
+
+            return true;
+        }
+        return false;
+    }
+
     private TestUtils() { }
 }

transport/src/main/java/io/netty/channel/socket/SctpData.java

@@ -1,121 +0,0 @@
-/*
- * Copyright 2011 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.socket;
-
-import com.sun.nio.sctp.MessageInfo;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufUtil;
-import io.netty.buffer.Unpooled;
-
-/**
- * Representation of SCTP Data Chunk
- */
-public final class SctpData implements SctpMessage {
-    private final int streamIdentifier;
-    private final int protocolIdentifier;
-
-    private final ByteBuf payloadBuffer;
-
-    private final MessageInfo msgInfo;
-
-    /**
-     * Essential data that is being carried within SCTP Data Chunk
-     * @param protocolIdentifier of payload
-     * @param streamIdentifier that you want to send the payload
-     * @param payloadBuffer channel buffer
-     */
-    public SctpData(int protocolIdentifier, int streamIdentifier, ByteBuf payloadBuffer) {
-        this.protocolIdentifier = protocolIdentifier;
-        this.streamIdentifier = streamIdentifier;
-        this.payloadBuffer = payloadBuffer;
-        this.msgInfo = null;
-    }
-
-    public SctpData(MessageInfo msgInfo, ByteBuf payloadBuffer) {
-        this.msgInfo = msgInfo;
-        this.streamIdentifier = msgInfo.streamNumber();
-        this.protocolIdentifier = msgInfo.payloadProtocolID();
-        this.payloadBuffer = payloadBuffer;
-    }
-
-    public int getStreamIdentifier() {
-        return streamIdentifier;
-    }
-
-    public int getProtocolIdentifier() {
-        return protocolIdentifier;
-    }
-
-    public ByteBuf getPayloadBuffer() {
-        if (payloadBuffer.readable()) {
-            return payloadBuffer.slice();
-        } else {
-            return Unpooled.EMPTY_BUFFER;
-        }
-    }
-
-    public MessageInfo getMessageInfo() {
-        return msgInfo;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-
-        SctpData sctpFrame = (SctpData) o;
-
-        if (protocolIdentifier != sctpFrame.protocolIdentifier) {
-            return false;
-        }
-
-        if (streamIdentifier != sctpFrame.streamIdentifier) {
-            return false;
-        }
-
-        if (!payloadBuffer.equals(sctpFrame.payloadBuffer)) {
-            return false;
-        }
-
-        return true;
-    }
-
-    @Override
-    public int hashCode() {
-        int result = streamIdentifier;
-        result = 31 * result + protocolIdentifier;
-        result = 31 * result + payloadBuffer.hashCode();
-        return result;
-    }
-
-    @Override
-    public String toString() {
-        return new StringBuilder().
-                append("SctpFrame{").
-                append("streamIdentifier=").
-                append(streamIdentifier).
-                append(", protocolIdentifier=").
-                append(protocolIdentifier).
-                append(", payloadBuffer=").
-                append(ByteBufUtil.hexDump(getPayloadBuffer())).
-                append('}').toString();
-    }
-}

transport/src/main/java/io/netty/channel/socket/SctpMessage.java

@@ -15,8 +15,115 @@
  */
 package io.netty.channel.socket;
 
+import com.sun.nio.sctp.MessageInfo;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+
 /**
- * A marker interface for a SCTP/IP message
+ * Representation of SCTP Data Chunk
  */
-public interface SctpMessage {
+public final class SctpMessage {
+    private final int streamIdentifier;
+    private final int protocolIdentifier;
+
+    private final ByteBuf payloadBuffer;
+
+    private MessageInfo msgInfo;
+
+    /**
+     * Essential data that is being carried within SCTP Data Chunk
+     * @param protocolIdentifier of payload
+     * @param streamIdentifier that you want to send the payload
+     * @param payloadBuffer channel buffer
+     */
+    public SctpMessage(int protocolIdentifier, int streamIdentifier, ByteBuf payloadBuffer) {
+        this.protocolIdentifier = protocolIdentifier;
+        this.streamIdentifier = streamIdentifier;
+        this.payloadBuffer = payloadBuffer;
+    }
+
+    public SctpMessage(MessageInfo msgInfo, ByteBuf payloadBuffer) {
+        this.msgInfo = msgInfo;
+        this.streamIdentifier = msgInfo.streamNumber();
+        this.protocolIdentifier = msgInfo.payloadProtocolID();
+        this.payloadBuffer = payloadBuffer;
+    }
+
+    public int getStreamIdentifier() {
+        return streamIdentifier;
+    }
+
+    public int getProtocolIdentifier() {
+        return protocolIdentifier;
+    }
+
+    public ByteBuf getPayloadBuffer() {
+        if (payloadBuffer.readable()) {
+            return payloadBuffer.slice();
+        } else {
+            return Unpooled.EMPTY_BUFFER;
+        }
+    }
+
+    public MessageInfo getMessageInfo() {
+        return msgInfo;
+    }
+
+    public boolean isComplete() {
+        if (msgInfo != null) {
+            return msgInfo.isComplete();
+        }  else {
+            //all outbound sctp messages are complete
+            return true;
+        }
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        SctpMessage sctpFrame = (SctpMessage) o;
+
+        if (protocolIdentifier != sctpFrame.protocolIdentifier) {
+            return false;
+        }
+
+        if (streamIdentifier != sctpFrame.streamIdentifier) {
+            return false;
+        }
+
+        if (!payloadBuffer.equals(sctpFrame.payloadBuffer)) {
+            return false;
+        }
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = streamIdentifier;
+        result = 31 * result + protocolIdentifier;
+        result = 31 * result + payloadBuffer.hashCode();
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return new StringBuilder().
+                append("SctpFrame{").
+                append("streamIdentifier=").
+                append(streamIdentifier).
+                append(", protocolIdentifier=").
+                append(protocolIdentifier).
+                append(", payloadBuffer=").
+                append(ByteBufUtil.hexDump(getPayloadBuffer())).
+                append('}').toString();
+    }
 }

transport/src/main/java/io/netty/channel/socket/SctpNotificationEvent.java

@@ -17,12 +17,12 @@ package io.netty.channel.socket;
 
 import com.sun.nio.sctp.Notification;
 
-public final class SctpNotification implements SctpMessage {
+public final class SctpNotificationEvent {
-    private final Notification notification;
+    private Notification notification;
-    private final Object attachment;
+    private Object attachment;
 
 
-    public SctpNotification(Notification notification, Object attachment) {
+    public SctpNotificationEvent(Notification notification, Object attachment) {
         this.notification = notification;
         this.attachment = attachment;
     }
@@ -35,7 +35,7 @@ public final class SctpNotification implements SctpMessage {
     }
 
     /**
-     * Return the attachment of this {@link SctpNotification}, or
+     * Return the attachment of this {@link SctpNotificationEvent}, or
      * <code>null</code> if no attachment was provided
      */
     public Object attachment() {
@@ -51,7 +51,7 @@ public final class SctpNotification implements SctpMessage {
             return false;
         }
 
-        SctpNotification that = (SctpNotification) o;
+        SctpNotificationEvent that = (SctpNotificationEvent) o;
 
         if (!attachment.equals(that.attachment)) {
             return false;

transport/src/main/java/io/netty/channel/socket/SctpNotificationHandler.java

@@ -57,7 +57,7 @@ public class SctpNotificationHandler extends AbstractNotificationHandler<Object>
     }
 
     private void updateInboundBuffer(Notification notification, Object o) {
-        sctpChannel.pipeline().inboundMessageBuffer().add(new SctpNotification(notification, o));
+        sctpChannel.pipeline().fireUserEventTriggered(new SctpNotificationEvent(notification, o));
     }
 }
 

transport/src/main/java/io/netty/channel/socket/nio/NioSctpChannel.java

@@ -29,7 +29,7 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.socket.DefaultSctpChannelConfig;
 import io.netty.channel.socket.SctpChannelConfig;
-import io.netty.channel.socket.SctpData;
+import io.netty.channel.socket.SctpMessage;
 import io.netty.channel.socket.SctpNotificationHandler;
 import io.netty.logging.InternalLogger;
 import io.netty.logging.InternalLoggerFactory;
@@ -229,13 +229,13 @@ public class NioSctpChannel extends AbstractNioMessageChannel implements io.nett
         }
 
         data.flip();
-        buf.add(new SctpData(messageInfo, Unpooled.wrappedBuffer(data)));
+        buf.add(new SctpMessage(messageInfo, Unpooled.wrappedBuffer(data)));
         return 1;
     }
 
     @Override
     protected int doWriteMessages(MessageBuf<Object> buf, boolean lastSpin) throws Exception {
-        SctpData packet = (SctpData) buf.peek();
+        SctpMessage packet = (SctpMessage) buf.peek();
         ByteBuf data = packet.getPayloadBuffer();
         int dataLen = data.readableBytes();
         ByteBuffer nioData;

transport/src/main/java/io/netty/channel/socket/oio/OioSctpChannel.java

@@ -29,7 +29,7 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.socket.DefaultSctpChannelConfig;
 import io.netty.channel.socket.SctpChannelConfig;
-import io.netty.channel.socket.SctpData;
+import io.netty.channel.socket.SctpMessage;
 import io.netty.channel.socket.SctpNotificationHandler;
 import io.netty.logging.InternalLogger;
 import io.netty.logging.InternalLoggerFactory;
@@ -120,7 +120,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
         }
 
         data.flip();
-        buf.add(new SctpData(messageInfo, Unpooled.wrappedBuffer(data)));
+        buf.add(new SctpMessage(messageInfo, Unpooled.wrappedBuffer(data)));
 
         if (readSuspended) {
             return 0;
@@ -132,7 +132,7 @@ public class OioSctpChannel extends AbstractOioMessageChannel
 
     @Override
     protected void doWriteMessages(MessageBuf<Object> buf) throws Exception {
-        SctpData packet = (SctpData) buf.poll();
+        SctpMessage packet = (SctpMessage) buf.poll();
         ByteBuf data = packet.getPayloadBuffer();
         int dataLen = data.readableBytes();
         ByteBuffer nioData;