transport/transport-http/src/main/java/org/opendaylight/netconf/transport/http/ClientHttp1RequestDispatcher.java

   1 /*
   2  * Copyright (c) 2024 PANTHEON.tech s.r.o. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.netconf.transport.http;
   9
  10 import com.google.common.util.concurrent.ListenableFuture;
  11 import com.google.common.util.concurrent.SettableFuture;
  12 import io.netty.channel.Channel;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.SimpleChannelInboundHandler;
  15 import io.netty.handler.codec.http.FullHttpRequest;
  16 import io.netty.handler.codec.http.FullHttpResponse;
  17 import java.util.Queue;
  18 import java.util.concurrent.ConcurrentLinkedQueue;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21
  22 /**
  23  * Client side {@link RequestDispatcher} implementation for HTTP 1.1.
  24  *
  25  * <p>
  26  * Serves as gateway to Netty {@link Channel}, performs sending requests to server, returns server responses associated.
  27  * Uses request to response mapping via queue -- first accepted response is associated with first request sent.
  28  */
  29 class ClientHttp1RequestDispatcher extends SimpleChannelInboundHandler<FullHttpResponse> implements RequestDispatcher {
  30     private static final Logger LOG = LoggerFactory.getLogger(ClientHttp1RequestDispatcher.class);
  31
  32     private final Queue<SettableFuture<FullHttpResponse>> queue = new ConcurrentLinkedQueue<>();
  33     private Channel channel = null;
  34
  35     ClientHttp1RequestDispatcher() {
  36         super(true); // auto-release
  37     }
  38
  39     @Override
  40     public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
  41         channel = ctx.channel();
  42         super.handlerAdded(ctx);
  43     }
  44
  45     @Override
  46     public ListenableFuture<FullHttpResponse> dispatch(final FullHttpRequest request) {
  47         if (channel == null) {
  48             throw new IllegalStateException("Connection is not established yet");
  49         }
  50         final var future = SettableFuture.<FullHttpResponse>create();
  51         channel.writeAndFlush(request).addListener(sent -> {
  52             final var cause = sent.cause();
  53             if (cause == null) {
  54                 queue.add(future);
  55             } else {
  56                 future.setException(cause);
  57             }
  58         });
  59         return future;
  60     }
  61
  62     @Override
  63     protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpResponse response) {
  64         final var future = queue.poll();
  65         if (future == null) {
  66             LOG.warn("Unexpected response while no future associated -- Dropping response object {}", response);
  67             return;
  68         }
  69
  70         if (!future.isDone()) {
  71             // NB using response' copy to disconnect the content data from channel's buffer allocated.
  72             // this prevents the content data became inaccessible once byte buffer of original message is released
  73             // on exit of current method
  74             future.set(response.copy());
  75         } else {
  76             LOG.warn("Future is already in Done state -- Dropping response object {}", response);
  77         }
  78     }
  79 }