【Quarkus】使用Buffer直接操作响应流-526互联


import io.vertx.core.buffer.Buffer;
import io.vertx.core.buffer.impl.BufferImpl;
import io.vertx.core.http.HttpHeaders;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.ext.web.RoutingContext;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
import jakarta.ws.rs.core.MediaType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Path("/api")
@Produces(MediaType.APPLICATION_JSON)
@Consumes(MediaType.APPLICATION_JSON)
public class StreamResource {

    private static final Logger log = LoggerFactory.getLogger(StreamResource.class);

    @GET
    @Path("/log")
    @Produces(MediaType.APPLICATION_OCTET_STREAM)
    public void getLog(@Context RoutingContext routingContext) throws IOException {
        HttpServerResponse response = routingContext.response();
        Buffer buffer = Buffer.buffer();
        buffer.getByteBuf().clear();
        OutputStream os = new OutputStream() {
            @Override
            public void write(int b) {
                checkShouldWriteToResponse();
                buffer.appendInt(b);
            }

            @Override
            public void write(byte[] b) {
                checkShouldWriteToResponse();
                buffer.appendBytes(b);
            }

            @Override
            public void close() {
                response.send(buffer);
            }

            private void checkShouldWriteToResponse() {
                if (buffer.length() < 10240) {
                    return;
                }
                log.debug("缓冲区满,输出");
                response.write(buffer);
                log.debug("输出完成,剩余: {}", buffer.length());
                ((BufferImpl) buffer).byteBuf().clear();
                log.debug("清理完成,剩余: {}", buffer.length());
            }
        };


        if (!response.headWritten()) {
            log.debug("填充响应头");
            response.putHeader(HttpHeaders.CONNECTION, HttpHeaders.CLOSE);
            response.putHeader(HttpHeaders.TRANSFER_ENCODING, HttpHeaders.CHUNKED);
        }

        List<String> size = routingContext.queryParam("size");
        var count = size.isEmpty() ? 10L : Long.valueOf(size.get(0));
        try {
            for (int i = 0; i < count; i++) {
                byte[] bytes = ("hello" + i + "\n").getBytes(StandardCharsets.UTF_8);
                os.write(bytes);
            }
        } finally {
            os.close();
        }

    }

}