java - Java 中带有 cachedHostConnectionPool 的 Source.queue

Question

我正在尝试在 java 中将 source.queue 与 cachedHostConnectionPool 一起使用，但无法获得正确的输出。下面是我的尝试。我总是得到成功的回应，我理解为什么，但不确定什么是正确的方法。

我得到成功响应的原因是因为我已将 Try 定义为 HttpResponse.create() ，它总是给我成功响应，但我怀疑为什么在我执行 queue.offer(Pair.create(httprequest) 时它没有被覆盖，承诺））。有人可以帮我理解我在这里做错了什么。

// Method to return queue 
  public SourceQueueWithComplete<Object> httpGetRequestWithPool
          (Materializer materializer) 
{
    final Http http = Http.get(system);
     final Flow<Pair<HttpRequest, Object>, Pair<Try<HttpResponse>, Object>, HostConnectionPool> flow;
      flow = http.cachedHostConnectionPool(ConnectHttp.toHost("host:port"));
    final SourceQueueWithComplete<Object> queue= Source.queue(3,OverflowStrategy.dropNew())
                 .map(d -> Pair.create(HttpRequest.create("contextpath"),d))
                 .via(flow)
                 .toMat(Sink.foreach(p -> p.first()), Keep.left())
                 .run(materializer);
    return queue;
  }

// Here i am trying to use.

  SourceQueueWithComplete<Object> queue = util.httpGetRequestWithPool(materializer);
        Source<Object, NotUsed> source = Source.from(ListOfObject);
        source.mapAsyncUnordered(3,x-> {
            String url = util.getServiceUrl("servicename").append(x.getId).toString();
            HttpRequest httpRequest = HttpRequest.create(url);
            Try<HttpResponse> promise = Try.apply(()->HttpResponse.create());
            queue.offer(Pair.create(httpRequest,promise))
                    .thenCompose(result -> {
                        if(result instanceof QueueOfferResult.Enqueued$){
                         return CompletableFuture.completedFuture(promise)
                                    .thenApply(res ->{
                                       if(res.get().status().intValue()==200){
                                           System.out.println("success");
                                       }
                                       return res;
                                    });
                        }
                        else{
                            return CompletableFuture.completedFuture(HttpResponse.create().withStatus(StatusCodes.SERVICE_UNAVAILABLE));

                        }
                    });
            return null;
        }).run(materializer);
```

标签： javaakka-http