java - 如何为 Java 11 HttpRequest 创建自定义 BodyPublisher

Question

我正在尝试创建一个自定义 BodyPublisher这将反序列化我的 JSON 对象。我可以在创建请求时反序列化 JSON 并使用 ofByteArray BodyPublishers的方法但我宁愿使用自定义发布者。

public class CustomPublisher implements HttpRequest.BodyPublisher {
    private byte[] bytes;
    
    public CustomPublisher(ObjectNode jsonData) {
        ...
        // Deserialize jsonData to bytes
        ...
    }
    
    @Override
    public long contentLength() {
        if(bytes == null) return 0;
        return bytes.length
    }
    
    @Override
    public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
        CustomSubscription subscription = new CustomSubscription(subscriber, bytes);
        subscriber.onSubscribe(subscription);       
    }

    private CustomSubscription implements Flow.Subscription {
         private final Flow.Subscriber<? super ByteBuffer> subscriber;
         private boolean cancelled;
         private Iterator<Byte> byterator;

         private CustomSubscription(Flow.Subscriber<? super ByteBuffer> subscriber, byte[] bytes) {
             this.subscriber = subscriber;
             this.cancelled = false;
             List<Byte> bytelist = new ArrayList<>();
             for(byte b : bytes) {
                 bytelist.add(b);
             }
             this.byterator = bytelist.iterator();
         }

         @Override
         public void request(long n) {
             if(cancelled) return;
             if(n < 0) {
                 subscriber.onError(new IllegalArgumentException());
             } else if(byterator.hasNext()) {
                 subscriber.onNext(ByteBuffer.wrap(new byte[]{byterator.next()));
             } else {
                 subscriber.onComplete();
             }
         }

         @Override
         public void cancel() {
             this.cancelled = true;
         }
    }
}

此实现有效，但前提是订阅 request方法以 1 作为参数被调用。但这就是我将它与 HttpRequest 一起使用时会发生的情况。
我很确定这不是创建自定义订阅的任何首选或最佳方式，但我还没有找到更好的方法来使它工作。
如果有人能引导我走上更好的道路，我将不胜感激。

Answer 1

避免使用字节数组是正确的，因为这会为大对象造成内存问题。
我不会尝试编写自定义发布者。相反，只需利用工厂方法 HttpRequest.BodyPublishers.ofInputStream .

HttpRequest.BodyPublisher publisher =
    HttpRequest.BodyPublishers.ofInputStream(() ->  {
        PipedInputStream in = new PipedInputStream();

        ForkJoinPool.commonPool().submit(() -> {
            try (PipedOutputStream out = new PipedOutputStream(in)) {
                objectMapper.writeTree(
                    objectMapper.getFactory().createGenerator(out),
                    jsonData);
            }
            return null;
        });

        return in;
    });

正如您所指出的，您可以使用 HttpRequest.BodyPublishers.ofByteArray .这对于相对较小的对象来说很好，但我出于习惯编程可扩展性。假设代码不需要扩展的问题是其他开发人员会假设传递大对象是安全的，而没有意识到对性能的影响。
编写您自己的正文发布者将需要大量工作。它的 subscribe方法继承自 Flow.Publisher . subscribe 的文档方法是这样开始的:

Adds the given Subscriber if possible.

每次您的 subscribe方法被调用，您需要将订阅者添加到某种集合中，您需要创建 Flow.Subscription 的实现，并且需要立即将其传递给订阅者的 onSubscribe方法。你的 Subscription 实现对象需要发回一个或多个 ByteBuffers，只有当 Subscription 的 request通过调用相应的订阅者(不仅仅是任何订阅者)来调用方法 onNext方法，一旦你发送了所有的数据，你必须调用同一个订阅者的 onComplete()方法。最重要的是，订阅实现对象需要处理 cancel要求。
您可以通过扩展 SubmissionPublisher 使这件事变得更容易。 ，这是 Flow.Publisher 的默认实现，然后添加一个 contentLength()方法。但正如 SubmissionPublisher 文档所示，即使是最小的工作实现，您仍有大量工作要做。
HttpRequest.BodyPublishers.of... 方法将为您完成所有这些工作。 ofByteArray小 object 没问题，但 ofInputStream将适用于您可以传入的任何对象。