Rx Netty：异步客户端不断收到“内容流已处理”错误

Question

在下面的代码中，我试图使用Rx Netty构建一个简单的异步HTTP客户端，它向api发出75个POST请求。我遇到的问题是，我不断收到“java.lang.IllegalStateException：内容流已被丢弃。”错误。

我在做什么错在这里？这是否与此：https://github.com/ReactiveX/RxNetty/issues/264

NioEventLoopGroup provider = new NioEventLoopGroup(); 

HttpClient<ByteBuf, ByteBuf> client = new HttpClientBuilder<ByteBuf, ByteBuf>("my-api.com", 80) 
    .eventloop(provider) 
    .build(); 

Gson gson = new Gson(); 

Observable.range(1, 75, Schedulers.from(provider)) 
    .flatMap(count -> { 
     Data data = new Data("test" + count); 

     return client.submit(
      HttpClientRequest.createPost("/create") 
       .withHeader("Authorization", AUTH_HEADER) 
       .withHeader("Content-Type", "application/json") 
       .withContent(gson.toJson(data)) 
    ); 
    }) 
    .flatMap(response -> { 
     return response.getContent().map((ByteBuf content) -> { 
     return gson.fromJson(content.toString(Charset.defaultCharset()), OtherData.class); 
     }); 
    }) 
    .subscribe(
     data -> logger.info("item done"), 
     err -> { 
      logger.error("Error", err); 

      provider.shutdownGracefully(); 
     }, 
     () -> { 
      logger.info("done"); 
      provider.shutdownGracefully(); 
     } 
    ); 

Answer 1

我已经意识到，当你做response.getContent()它需要一个地图，而不是一个flatMap里面做。例如：

NioEventLoopGroup provider = new NioEventLoopGroup(); 

HttpClient<ByteBuf, ByteBuf> client = new HttpClientBuilder<ByteBuf, 
ByteBuf>("my-api.com", 80) 
.eventloop(provider) 
.build(); 

Gson gson = new Gson(); 

Observable.range(1, 75, Schedulers.from(provider)) 
.flatMap(count -> { 
    Data data = new Data("test" + count); 

    return client.submit(
     HttpClientRequest.createPost("/create") 
      .withHeader("Authorization", AUTH_HEADER) 
      .withHeader("Content-Type", "application/json") 
      .withContent(gson.toJson(data)) 
); 
}) 
// map not flatMap 
.map(response -> { 
    return response.getContent().map((ByteBuf content) -> { 
    return gson.fromJson(content.toString(Charset.defaultCharset()), OtherData.class); 
    }); 
}) 
.subscribe(
    data -> logger.info("item done"), 
    err -> { 
     logger.error("Error", err); 

     provider.shutdownGracefully(); 
    }, 
    () -> { 
     logger.info("done"); 
     provider.shutdownGracefully(); 
    } 
);