Добавьте многоканальные конвейерные обработчики в SwiftNIO аналогично Java Netty

Question

Я изучаю, как добавить несколько обработчиков в канальные конвейеры в SwiftNIO.В Java Netty у меня есть следующий код:

@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);

    }
} 

В SwiftNIO, похоже, нет подобных классов, как "LengthFieldBasedFrameDecoder", "ProtobufDecoder", "LengthFieldPrepender", "ProtobufEncoder".Как я могу получить те в SwiftNIO?

Answer 1

Хорошо, позвольте мне пройтись по всем обработчикам, которые вы добавляете в свой конвейер в Netty:

• IdleStateHandler: доступно с import NIO из пакета swift-nio
• LengthFieldBasedFrameDecoder: прямо сейчас в PR , но будет доступно через import NIOExtras из пакета swift-nio-extras
• ProtobufDecoder, LengthFieldPrepender,ProtobufEncoder: все в настоящее время недоступно, но реализовать просто:

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)
            }
        }
    }