class PipePayloadParser implements PayloadParser {

    private final static AtomicIntegerFieldUpdater<PipePayloadParser> CURRENT_PIPE =
        AtomicIntegerFieldUpdater.newUpdater(PipePayloadParser.class, "currentPipe");

    private final Sinks.Many<Buffer> sink = Reactors.createMany();

    private final List<BiConsumer<Buffer, PipePayloadParser>> pipe = new CopyOnWriteArrayList<>();

    private final List<Buffer> result = new CopyOnWriteArrayList<>();

    private volatile RecordParser recordParser;

    private Function<Buffer, Buffer> directMapper;

    private Consumer<RecordParser> firstInit;

    private volatile int currentPipe;

    public Buffer newBuffer() {
        return Buffer.buffer();
    }

    public PipePayloadParser result(String buffer) {
        return result(Buffer.buffer(buffer));
    }

    public PipePayloadParser result(byte[] buffer) {
        return result(Buffer.buffer(buffer));
    }

//    public PipePayloadParser handler(Consumer<Buffer> handler) {
//
//        return handler((payloadParser, buffer) -> handler.accept(buffer));
//    }

    public PipePayloadParser handler(BiConsumer<Buffer, PipePayloadParser> handler) {
        pipe.add(handler);
        return this;
    }


    public PipePayloadParser delimited(String delimited) {
        if (recordParser == null) {
            setParser(RecordParser.newDelimited(delimited));
            firstInit = (parser -> parser.delimitedMode(delimited));
            return this;
        }
        recordParser.delimitedMode(delimited);
        return this;
    }

    public PipePayloadParser fixed(int size) {
        if (size == 0) {
            complete();
            return this;
        }
        if (recordParser == null) {
            setParser(RecordParser.newFixed(size));
            firstInit = (parser -> parser.fixedSizeMode(size));
            return this;
        }
        recordParser.fixedSizeMode(size);
        return this;
    }

    public PipePayloadParser direct(Function<Buffer, Buffer> mapper) {
        this.directMapper = mapper;
        return this;
    }

    private BiConsumer<Buffer, PipePayloadParser> getNextHandler() {
        int i = CURRENT_PIPE.getAndIncrement(this);
        if (i < pipe.size()) {
            return pipe.get(i);
        }
        CURRENT_PIPE.set(this, 0);
        return pipe.get(0);
    }

    private void setParser(RecordParser parser) {
        this.recordParser = parser;
        this.recordParser.handler(buffer -> getNextHandler().accept(buffer, this));
    }

    public PipePayloadParser complete() {<FILL_FUNCTION_BODY>}

    public PipePayloadParser result(Buffer buffer) {
        this.result.add(buffer);
        return this;
    }

    @Override
    public synchronized void handle(Buffer buffer) {
        if (recordParser == null && directMapper == null) {
            log.error("record parser not init");
            return;
        }
        if (recordParser != null) {
            recordParser.handle(buffer);
            return;
        }
        Buffer buf = directMapper.apply(buffer);
        if (null != buf) {
            sink.emitNext(buf, Reactors.emitFailureHandler());
        }
    }

    @Override
    public Flux<Buffer> handlePayload() {
        return sink.asFlux();
    }

    @Override
    public void reset() {
        this.result.clear();
        complete();
    }

    @Override
    public void close() {
        sink.tryEmitComplete();
        CURRENT_PIPE.set(this, 0);
        this.result.clear();
    }

}

        CURRENT_PIPE.set(this, 0);
        if (recordParser != null) {
            firstInit.accept(recordParser);
        }
        if (!this.result.isEmpty()) {
            Buffer buffer = Buffer.buffer();
            for (Buffer buf : this.result) {
                buffer.appendBuffer(buf);
            }
            this.result.clear();
            sink.emitNext(buffer, Reactors.emitFailureHandler());
        }
        return this;

    
<no_super_class>