Add multiple channel pipeline handlers in SwiftNIO similarly as Java Netty

Question

I am exploring how to add multiple handlers in channel pipelines in SwiftNIO. In Java Netty, I have the following code:

@Component
public class NettyClientFilter extends ChannelInitializer<SocketChannel> {

    @Autowired
    private NettyClientHandler nettyClientHandler;

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline ph = ch.pipeline();

        ph.addLast(new IdleStateHandler(20, 10, 0));
        ph.addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));

        ph.addLast(new ProtobufDecoder(IMessage.getDefaultInstance()));

        ph.addLast(new LengthFieldPrepender(4));
        ph.addLast(new ProtobufEncoder());
        ph.addLast("nettyClientHandler",nettyClientHandler);

    }
} 

In the SwiftNIO, seems there are no similar classes as "LengthFieldBasedFrameDecoder", "ProtobufDecoder", "LengthFieldPrepender", "ProtobufEncoder". How can I get those ones in SwiftNIO?

Answer 1

Right, let me go through all the handlers you add to your pipeline in Netty:

• IdleStateHandler: available with import NIO from the swift-nio package
• LengthFieldBasedFrameDecoder: right now in a PR but will be available shortly through import NIOExtras from the swift-nio-extras package
• ProtobufDecoder, LengthFieldPrepender, ProtobufEncoder: all currently unavailable but straightforward to implement:

LengthFieldPrepender:

    final class LengthFieldPrepender<IntType: FixedWidthInteger>: ChannelOutboundHandler {
        // we send send and receive ByteBuffers
        typealias OutboundIn = ByteBuffer
        typealias OutboundOut = ByteBuffer

        private let endianness: Endianness
        private var buf: ByteBuffer?

        init(type: IntType.Type = IntType.self, endianness: Endianness = .big) {
            self.endianness = endianness
        }

        func handlerAdded(ctx: ChannelHandlerContext) {
            self.buf = ctx.channel.allocator.buffer(capacity: 8)
        }

        func write(ctx: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
            let incomingData = self.unwrapOutboundIn(data)

            // we cache `self.buf` so we might get lucky and save an allocation here if the previous buffer has been fully written already
            self.buf!.clear()
            // write the length as the right type
            self.buf!.write(integer: IntType(incomingData.readableBytes), endianness: self.endianness)
            ctx.write(self.wrapOutboundOut(self.buf!), promise: nil)
            // write the actual data
            ctx.write(data, promise: promise)
        }
    }

ProtobufDecoder:

    import SwiftProtobuf
    import NIOFoundationCompat // for ByteBuffer.readData

    final class ProtobufDecoder<Msg: SwiftProtobuf.Message>: ChannelInboundHandler {
        typealias InboundIn = ByteBuffer
        typealias InboundOut = Msg

        func channelRead(ctx: ChannelHandlerContext, data: NIOAny) {
            var buffer = self.unwrapInboundIn(data)
            let data = buffer.readData(length: buffer.readableBytes)!
            do {
                // pretty straightforward here, just call the message type's initialiser
                let req = try Msg(serializedData: data)
                ctx.fireChannelRead(self.wrapInboundOut(req))
            } catch {
                ctx.fireErrorCaught(error)
            }
        }
    }

ProtobufEncoder:

    import NIOFoundationCompat
    import SwiftProtobuf

    final class ProtobufEncoder<Msg: SwiftProtobuf.Message>: ChannelOutboundHandler {
        typealias OutboundIn = Msg
        typealias OutboundOut = ByteBuffer

        private var buf: ByteBuffer?

        func handlerAdded(ctx: ChannelHandlerContext) {
            self.buf = ctx.channel.allocator.buffer(capacity: 4096)
        }

        func write(ctx: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
            let msg = self.unwrapOutboundIn(data)
            self.buf!.clear()
            do {
                // just use SwiftProtobuf's nice encoder
                self.buf!.write(bytes: try msg.serializedData())
                ctx.write(self.wrapOutboundOut(self.buf!), promise: promise)
            } catch {
                ctx.fireErrorCaught(error)
            }
        }
    }