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.FutureCallback;
  14 import io.netty.channel.Channel;
  15 import io.netty.channel.ChannelHandlerContext;
  16 import io.netty.handler.codec.http.FullHttpRequest;
  17 import io.netty.handler.codec.http.FullHttpResponse;
  18 import io.netty.handler.codec.http.HttpScheme;
  19 import io.netty.handler.ssl.SslHandler;
  20 import java.util.Map;
  21 import java.util.concurrent.ConcurrentHashMap;
  22 import java.util.concurrent.atomic.AtomicInteger;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 /**
  27  * Client side {@link RequestDispatcher} implementation for HTTP 2.
  28  *
  29  * <p>
  30  * Serves as gateway to Netty {@link Channel}, performs sending requests to server, returns server responses associated.
  31  * Uses request to response mapping by stream identifier.
  32  */
  33 final class ClientHttp2RequestDispatcher extends ClientRequestDispatcher {
  34     private static final Logger LOG = LoggerFactory.getLogger(ClientHttp2RequestDispatcher.class);
  35
  36     // TODO: we access the queue only from Netty callbacks: can we use a plain HashMap?
  37     private final Map<Integer, FutureCallback<FullHttpResponse>> map = new ConcurrentHashMap<>();
  38     // identifier for streams initiated from client require to be odd-numbered, 1 is reserved
  39     // see https://datatracker.ietf.org/doc/html/rfc7540#section-5.1.1
  40     private final AtomicInteger streamIdCounter = new AtomicInteger(3);
  41
  42     private boolean ssl = false;
  43
  44     @Override
  45     public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
  46         ssl = ctx.pipeline().get(SslHandler.class) != null;
  47         super.handlerAdded(ctx);
  48     }
  49
  50     @Override
  51     public void dispatch(final Channel channel, final FullHttpRequest request,
  52             final FutureCallback<FullHttpResponse> callback) {
  53         final var streamId = streamIdCounter.getAndAdd(2);
  54         request.headers()
  55             .setInt(STREAM_ID.text(), streamId)
  56             .set(SCHEME.text(), ssl ? HttpScheme.HTTPS.name() : HttpScheme.HTTP.name());
  57
  58         channel.writeAndFlush(request).addListener(sent -> {
  59             final var cause = sent.cause();
  60             if (cause == null) {
  61                 map.put(streamId, callback);
  62             } else {
  63                 callback.onFailure(cause);
  64             }
  65         });
  66     }
  67
  68     @Override
  69     protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpResponse response) {
  70         final var streamId = response.headers().getInt(STREAM_ID.text());
  71         if (streamId == null) {
  72             LOG.warn("Unexpected response with no stream ID -- Dropping response object {}", response);
  73             return;
  74         }
  75         final var callback = map.remove(streamId);
  76         if (callback != null) {
  77             callback.onSuccess(response);
  78         } else {
  79             LOG.warn("Unexpected response with unknown or expired stream ID {} -- Dropping response object {}",
  80                 streamId, response);
  81         }
  82     }
  83 }