transport/transport-http/src/main/java/org/opendaylight/netconf/transport/http/ClientHttp2RequestDispatcher.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 static io.netty.handler.codec.http2.HttpConversionUtil.ExtensionHeaderNames.SCHEME;
  11 import static io.netty.handler.codec.http2.HttpConversionUtil.ExtensionHeaderNames.STREAM_ID;
  12
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.SettableFuture;
  15 import io.netty.channel.Channel;
  16 import io.netty.channel.ChannelHandlerContext;
  17 import io.netty.channel.SimpleChannelInboundHandler;
  18 import io.netty.handler.codec.http.FullHttpRequest;
  19 import io.netty.handler.codec.http.FullHttpResponse;
  20 import io.netty.handler.codec.http.HttpScheme;
  21 import io.netty.handler.ssl.SslHandler;
  22 import java.util.Map;
  23 import java.util.concurrent.ConcurrentHashMap;
  24 import java.util.concurrent.atomic.AtomicInteger;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 /**
  29  * Client side {@link RequestDispatcher} implementation for HTTP 2.
  30  *
  31  * <p>
  32  * Serves as gateway to Netty {@link Channel}, performs sending requests to server, returns server responses associated.
  33  * Uses request to response mapping by stream identifier.
  34  */
  35 class ClientHttp2RequestDispatcher extends SimpleChannelInboundHandler<FullHttpResponse> implements RequestDispatcher {
  36     private static final Logger LOG = LoggerFactory.getLogger(ClientHttp2RequestDispatcher.class);
  37
  38     private final Map<Integer, SettableFuture<FullHttpResponse>> map = new ConcurrentHashMap<>();
  39     private final AtomicInteger streamIdCounter = new AtomicInteger(3);
  40
  41     private Channel channel = null;
  42     private boolean ssl = false;
  43
  44     ClientHttp2RequestDispatcher() {
  45         super(true); // auto-release
  46     }
  47
  48     private Integer nextStreamId() {
  49         // identifier for streams initiated from client require to be odd-numbered, 1 is reserved
  50         // see https://datatracker.ietf.org/doc/html/rfc7540#section-5.1.1
  51         return streamIdCounter.getAndAdd(2);
  52     }
  53
  54     @Override
  55     public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
  56         channel = ctx.channel();
  57         ssl = ctx.pipeline().get(SslHandler.class) != null;
  58         super.handlerAdded(ctx);
  59     }
  60
  61     @Override
  62     public ListenableFuture<FullHttpResponse> dispatch(final FullHttpRequest request) {
  63         if (channel == null) {
  64             throw new IllegalStateException("Connection is not established yet");
  65         }
  66         final var streamId = nextStreamId();
  67         request.headers().setInt(STREAM_ID.text(), streamId);
  68         request.headers().set(SCHEME.text(), ssl ? HttpScheme.HTTPS.name() : HttpScheme.HTTP.name());
  69
  70         final var future = SettableFuture.<FullHttpResponse>create();
  71         channel.writeAndFlush(request).addListener(sent -> {
  72             if (sent.cause() == null) {
  73                 map.put(streamId, future);
  74             } else {
  75                 future.setException(sent.cause());
  76             }
  77         });
  78         return future;
  79     }
  80
  81     @Override
  82     protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpResponse response) {
  83         final var streamId = response.headers().getInt(STREAM_ID.text());
  84         if (streamId == null) {
  85             LOG.warn("Unexpected response with no stream ID -- Dropping response object {}", response);
  86             return;
  87         }
  88         final var future = map.remove(streamId);
  89         if (future == null) {
  90             LOG.warn("Unexpected response with unknown or expired stream ID {} -- Dropping response object {}",
  91                 streamId, response);
  92             return;
  93         }
  94         if (!future.isDone()) {
  95             // NB using response' copy to disconnect the content data from channel's buffer allocated.
  96             // this prevents the content data became inaccessible once byte buffer of original message is released
  97             // on exit of current method
  98             future.set(response.copy());
  99         } else {
 100             LOG.warn("Future is already in Done state -- Dropping response object {}", response);
 101         }
 102     }
 103 }