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.ChannelInboundHandler;
import io.netty.channel.sctp.SctpMessage;
import io.netty.handler.codec.MessageToMessageDecoder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class SctpMessageCompletionHandler extends MessageToMessageDecoder<SctpMessage> {
private final Map<Integer, ByteBuf> fragments = new HashMap<Integer, ByteBuf>();
@Override
protected void decode(ChannelHandlerContext ctx, SctpMessage msg, List<Object> out) throws Exception {
final ByteBuf byteBuf = msg.content();
final int protocolIdentifier = msg.protocolIdentifier();
final int streamIdentifier = msg.streamIdentifier();
final boolean isComplete = msg.isComplete();
final boolean isUnordered = msg.isUnordered();
ByteBuf frag = fragments.remove(streamIdentifier);
if (frag == null) {
frag = Unpooled.EMPTY_BUFFER;
}
if (isComplete && !frag.isReadable()) {
out.add(msg);
} else if (!isComplete && frag.isReadable()) {
fragments.put(streamIdentifier, Unpooled.wrappedBuffer(frag, byteBuf));
} else if (isComplete && frag.isReadable()) {
SctpMessage assembledMsg = new SctpMessage(
protocolIdentifier,
streamIdentifier,
isUnordered,
Unpooled.wrappedBuffer(frag, byteBuf));
out.add(assembledMsg);
} else {
fragments.put(streamIdentifier, byteBuf);
}
byteBuf.retain();
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
for (ByteBuf buffer: fragments.values()) {
buffer.release();
}
fragments.clear();
super.handlerRemoved(ctx);
}
}