Netty HTTPクライアントでリクエストを再試行する

Question

NettyベースのHTTPクライアントでHTTPリクエストを再試行する方法はありますか？

は、HTTP応答コード503を受信した場合、1秒後にHTTP要求を再試行しようとする次のハンドラを、検討：

私はこの例では、チャネルへの書き込み

public class RetryChannelHandler extends ChannelDuplexHandler { 
    List<HttpObject> requestParts; 

    @Override 
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
     if (msg instanceof HttpRequest) { 
      requestParts = new ArrayList<>(); 
      requestParts.add((HttpRequest)msg); 
     } else if (msg instanceof HttpObject) { 
      requestParts.add((HttpObject)msg); 
     } 

     super.write(ctx, msg, promise); 
    } 

    @Override 
    public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception { 
     if (msg instanceof HttpResponse) { 
      HttpResponse res = (HttpResponse)msg; 
      if (res.status().code() == 503) { 
       ctx.executor().schedule(new Runnable() { 
        @Override 
        public void run() { 
         for (HttpObject obj : requestParts) { 
          ctx.channel().write(obj); 
         } 
        } 
       }, 1000, TimeUnit.MILLISECONDS); 
      } else { 
       super.channelRead(ctx, msg); 
      } 
     } else { 
      super.channelRead(ctx, msg); 
     } 
    } 
} 

、パイプライン内の他のハンドラHttpObjectsを参照してください。ただし、HttpRequestは実際には再度実行されません。ただ1つのHttpResponseしか受け取られません。

私はこのケースでは単にChannelを悪用していると思うし、再試行を実行するために新しいChannel（サーバーへの新しい接続を表す）を作成する必要があります。私には分かりませんが、ハンドラのコンテキストから新しいチャンネルを作成する方法と、この種のロジックを実行するNettyの適切な層に本当にいるかどうかです。

私が説明している種類の行動を達成する方法についてのガイダンスは高く評価されます。

Answer 1

write(...)を呼び出した後にflush()に電話する必要があります。それ以外の場合は、チャンネルにフラッシュされません。また、おそらく（）と複製（）HttpContentを確実に保持する必要があります。さもなければ、すでにリリースされたHttpContentオブジェクトを書き込もうとするかもしれません。このような

何か（テストしていません）：

public class RetryChannelHandler extends ChannelDuplexHandler { 
    Queue<HttpObject> requestParts; 

    @Override 
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
     if (msg instanceof HttpRequest) { 
      requestParts = new ArrayDeque<>(); 
      requestParts.add((HttpRequest)msg); 
     } else if (msg instanceof HttpContent) { 
      requestParts.add(((HttpContent)msg).duplicate().retain()); 
     } 

     super.write(ctx, msg, promise); 
    } 

    @Override 
    public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception { 
     if (msg instanceof HttpResponse) { 
      HttpResponse res = (HttpResponse)msg; 
      if (res.status().code() == 503) { 
       ctx.executor().schedule(new Runnable() { 
        @Override 
        public void run() { 
         HttpObject obj; 
         while ((obj = requestParts.poll()) != null) { 
          ctx.write(obj); 
         } 
         ctx.flush(); 
        } 
       }, 1000, TimeUnit.MILLISECONDS); 
      } else { 
       HttpObject obj; 
       while ((obj = requestParts.poll()) != null) { 
        ReferenceCountUtil.release(obj); 
       } 
       super.channelRead(ctx, msg); 
      } 
     } else { 
      super.channelRead(ctx, msg); 
     } 
    } 
}