Repository: apple/swift-nio
File: Sources/NIOPerformanceTester/TCPThroughputBenchmark.swift (Swift)

Code before:
            }
        }
    }

    public init(messages: Int, messageSize: Int) {
        self.messages = messages
        self.messageSize = messageSize
    }

    func setUp() throws {
        self.group = MultiThreadedEventLoopGroup(numberOfThreads: 4)

        let connectionEstablished: EventLoopPromise<Void> = self.group.next().makePromise()

        self.serverChannel = try ServerBootstrap(group: self.group)
            .childChannelInitializer { channel in
                self.serverHandler = ServerHandler()
                connectionEstablished.succeed()
                return channel.pipeline.addHandler(self.serverHandler)
            }
            .bind(host: "127.0.0.1", port: 0)
            .wait()

        self.clientChannel = try ClientBootstrap(group: group)
            .channelInitializer { channel in
                channel.pipeline.addHandler(ByteToMessageHandler(StreamDecoder())).flatMap { v in
                    channel.pipeline.addHandler(ClientHandler(self))
                }
            }
            .connect(to: serverChannel.localAddress!)
            .wait()

        var message = self.serverChannel.allocator.buffer(capacity: self.messageSize)
        message.writeInteger(UInt16(messageSize), as:UInt16.self)
        for idx in 0..<(self.messageSize - MemoryLayout<UInt16>.stride) {
            message.writeInteger(UInt8(idx%(Int(UInt8.max)+1)), endianness:.little, as:UInt8.self)
        }
        self.message = message

        try connectionEstablished.futureResult.wait()
    }

    func tearDown() {
        try! self.clientChannel.close().wait()
        try! self.serverChannel.close().wait()
        try! self.group.syncShutdownGracefully()
    }

    func run() throws -> Int {
        self.isDonePromise = self.group.next().makePromise()
        Task {

Review comment (suggestion):
```suggestion
                channel.pipeline.addHandler(ByteToMessageHandler(StreamDecoder())).flatMap { _ in
```

Code after:
        private let benchmark: TCPThroughputBenchmark
        private var messagesReceived: Int

        init(_ benchmark: TCPThroughputBenchmark) {
            self.benchmark = benchmark
            self.messagesReceived = 0
        }

        public func channelRead(context: ChannelHandlerContext, data: NIOAny) {
            self.messagesReceived += 1
            if (self.benchmark.messages == self.messagesReceived) {
                self.benchmark.isDonePromise.succeed()
                self.messagesReceived = 0
            }
        }
    }

    public init(messages: Int, messageSize: Int) {
        self.messages = messages
        self.messageSize = messageSize
    }

    func setUp() throws {
        self.group = MultiThreadedEventLoopGroup(numberOfThreads: 4)

        let connectionEstablished: EventLoopPromise<Void> = self.group.next().makePromise()

        self.serverChannel = try ServerBootstrap(group: self.group)
            .childChannelInitializer { channel in
                self.serverHandler = ServerHandler()
                connectionEstablished.succeed()
                return channel.pipeline.addHandler(self.serverHandler)
            }
            .bind(host: "127.0.0.1", port: 0)
            .wait()

        self.clientChannel = try ClientBootstrap(group: group)
            .channelInitializer { channel in
                channel.pipeline.addHandler(ByteToMessageHandler(StreamDecoder())).flatMap { _ in
                    channel.pipeline.addHandler(ClientHandler(self))
                }
            }
            .connect(to: serverChannel.localAddress!)
            .wait()

        var message = self.serverChannel.allocator.buffer(capacity: self.messageSize)
        message.writeInteger(UInt16(messageSize), as:UInt16.self)
        for idx in 0..<(self.messageSize - MemoryLayout<UInt16>.stride) {
            message.writeInteger(UInt8(truncatingIfNeeded: idx), endianness:.little, as:UInt8.self)
        }
        self.message = message