code: /*
   * Copyright (c) 2019 the Eclipse Milo Authors
   *
   * This program and the accompanying materials are made
   * available under the terms of the Eclipse Public License 2.0
   * which is available at https://www.eclipse.org/legal/epl-2.0/
   *
   * SPDX-License-Identifier: EPL-2.0
   */
  
  package org.eclipse.milo.opcua.stack.client.transport;
  
  import java.nio.channels.ClosedChannelException;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.TimeUnit;
  
  import io.netty.channel.Channel;
  import io.netty.util.HashedWheelTimer;
  import io.netty.util.Timeout;
  import org.eclipse.milo.opcua.stack.client.UaStackClient;
  import org.eclipse.milo.opcua.stack.client.UaStackClientConfig;
  import org.eclipse.milo.opcua.stack.core.StatusCodes;
  import org.eclipse.milo.opcua.stack.core.UaException;
  import org.eclipse.milo.opcua.stack.core.serialization.UaRequestMessage;
  import org.eclipse.milo.opcua.stack.core.serialization.UaResponseMessage;
  import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public abstract class AbstractTransport implements UaTransport {
  
      private final Logger logger = LoggerFactory.getLogger(getClass());
  
      private final HashedWheelTimer wheelTimer;
  
      private final UaStackClientConfig config;
  
      public AbstractTransport(UaStackClientConfig config) {
          this.config = config;
  
          wheelTimer = config.getWheelTimer();
      }
  
      /**
       * Get the {@link UaStackClientConfig} of the {@link UaStackClient} this transport instance belongs to.
       *
       * @return the {@link UaStackClientConfig} of the {@link UaStackClient} this transport instance belongs to.
       */
      public UaStackClientConfig getConfig() {
          return config;
      }
  
      /**
       * Get a {@link Channel} suitable for sending a request on.
       * <p>
       * The Channel must have a handler capable of matching inbound responses to pending outbound
       * {@link UaTransportRequest}s.
       *
       * @return a {@link Channel} suitable for sending a request on.
       */
      public abstract CompletableFuture<Channel> channel();
  
      @Override
      public CompletableFuture<UaResponseMessage> sendRequest(UaRequestMessage request) {
          return channel().thenCompose(channel -> sendRequest(request, channel, true));
      }
  
      protected CompletableFuture<UaResponseMessage> sendRequest(
          UaRequestMessage request,
          Channel channel,
          boolean firstAttempt) {
  
          UaTransportRequest transportRequest = new UaTransportRequest(request);
  
          scheduleRequestTimeout(transportRequest);
  
          transportRequest.getFuture().whenComplete(
              (response, ex) ->
                  cancelRequestTimeout(transportRequest)
          );
  
          channel.writeAndFlush(transportRequest).addListener(f -> {
              if (!f.isSuccess()) {
                  Throwable cause = f.cause();
  
                  if (cause instanceof ClosedChannelException && firstAttempt) {
                      logger.debug("Write failed, channel closed; retrying...");
  
                      config.getScheduledExecutor().schedule(
                          () -> config.getExecutor().execute(() -> {
                              CompletableFuture<UaResponseMessage> sendAgain =
                                  channel().thenCompose(ch -> sendRequest(request, ch, false));
  
                              sendAgain.whenComplete((r, ex) -> {
                                  if (r != null) {
                                      transportRequest.getFuture().complete(r);
                                  } else {
                                      transportRequest.getFuture().completeExceptionally(ex);
                                  }
                              });
                          }),
                          1,
                          TimeUnit.SECONDS
                      );
                  } else {
                      transportRequest.getFuture().completeExceptionally(cause);
  
                      logger.debug(
                          "Write failed, request={}, requestHandle={}",
                          request.getClass().getSimpleName(),
                          request.getRequestHeader().getRequestHandle());
                  }
              } else {
                  if (logger.isTraceEnabled()) {
                      logger.trace(
                          "Write succeeded for request={}, requestHandle={}",
                          request.getClass().getSimpleName(),
                          request.getRequestHeader().getRequestHandle());
                  }
              }
          });
  
          return transportRequest.getFuture();
      }
  
      private void scheduleRequestTimeout(UaTransportRequest transportRequest) {
          RequestHeader requestHeader = transportRequest.getRequest().getRequestHeader();
  
          long timeoutHint = requestHeader.getTimeoutHint() != null ?
              requestHeader.getTimeoutHint().longValue() : 0L;
  
          if (timeoutHint > 0) {
              Timeout timeout = wheelTimer.newTimeout(
                  t -> {
                      UaException exception = new UaException(
                          StatusCodes.Bad_Timeout,
                          String.format(
                              "requestId=%s timed out after %sms",
                              requestHeader.getRequestHandle(), timeoutHint)
                      );
  
                      transportRequest.getFuture().completeExceptionally(exception);
                  },
                  timeoutHint,
                  TimeUnit.MILLISECONDS
              );
  
              transportRequest.setTimeout(timeout);
          }
      }
  
      private void cancelRequestTimeout(UaTransportRequest transportRequest) {
          Timeout timeout = transportRequest.getTimeout();
          if (timeout != null) timeout.cancel();
      }
  
  }
  