public <T> Mono<T> sendRequest(RpcRequest<T> request, ByteBuf requestBody) {
        ChannelOperations<?, ?> conn = connection.getConnection();
        if (requestBody.refCnt() == 0) {
            return Mono.error(new IllegalArgumentException("The request body has been released"));
        }
        if (conn.isDisposed()) {
            requestBody.release();
            return Mono.error(new ClosedChannelException());
        }
        Sinks.One<T> sink = Sinks.one();
        while (true) {
            int requestId = generateRandomId();
            Sinks.One<?> previous = pendingRequestMap.putIfAbsent(requestId, sink);
            if (previous != null) {
                continue;
            }
            request.setRequestId(requestId);
            ByteBuf buffer;
            try {
                buffer = RpcFrameEncoder.INSTANCE.encodeRequest(request, requestBody);
            } catch (Exception e) {
                requestBody.release();
                resolveRequest(requestId,
                        null,
                        new SerializationException(
                                "Failed to encode the request: "
                                        + request,
                                e));
                break;
            }
            // sendObject() will release the buffer no matter it succeeds or fails

            // Duplicate the buffer to use an independent reader index
            // because we don't want to modify the reader index of the original buffer
            // if it is an unreleasable buffer internally, or it may be sent to multiple endpoints.
            // Note that the content of the buffer is not copied, so "duplicate()" is efficient.
            conn.sendObject(buffer.duplicate())
                    .then()
                    .subscribe(null, t -> resolveRequest(requestId, null, t));
            break;
        }
        return sink.asMono();
    }
Accept {@param requestBody} of ByteBuf so that we can send the same buffer to multiple peers

@implNote The method ensures {@param requestBody} will be released by 1
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/rpc/RpcEndpoint.java