class PortForwarderWebsocket {

  private static final Logger LOG = LoggerFactory.getLogger(PortForwarderWebsocket.class);

  private final HttpClient client;
  private final Executor executor;
  private final long connectTimeoutMills;

  public PortForwarderWebsocket(HttpClient client, Executor executor, long connectTimeoutMillis) {
    this.client = client;
    this.executor = executor;
    this.connectTimeoutMills = connectTimeoutMillis;
  }

  public LocalPortForward forward(final URL resourceBaseUrl, final int port, final InetAddress localHost, final int localPort) {<FILL_FUNCTION_BODY>}

  public PortForward forward(URL resourceBaseUrl, int port, final ReadableByteChannel in, final WritableByteChannel out) {
    final PortForwarderWebsocketListener listener = new PortForwarderWebsocketListener(in, out, executor);
    CompletableFuture<WebSocket> socket = client
        .newWebSocketBuilder()
        .uri(URI.create(URLUtils.join(resourceBaseUrl.toString(), "portforward?ports=" + port)))
        .connectTimeout(connectTimeoutMills, TimeUnit.MILLISECONDS)
        .subprotocol("v4.channel.k8s.io")
        .buildAsync(listener);

    socket.whenComplete((w, t) -> {
      if (t != null) {
        listener.onError(w, t);
      }
    });

    return new PortForward() {
      private final AtomicBoolean closed = new AtomicBoolean();

      @Override
      public void close() {
        if (!closed.compareAndSet(false, true)) {
          return;
        }
        socket.whenComplete((w, t) -> {
          if (w != null) {
            listener.closeBothWays(w, 1001, "User closing");
          }
        });
      }

      @Override
      public boolean isAlive() {
        return listener.isAlive();
      }

      @Override
      public boolean errorOccurred() {
        return listener.errorOccurred();
      }

      @Override
      public Collection<Throwable> getClientThrowables() {
        return listener.getClientThrowables();
      }

      @Override
      public Collection<Throwable> getServerThrowables() {
        return listener.getServerThrowables();
      }
    };
  }

  InetSocketAddress createNewInetSocketAddress(InetAddress localHost, int localPort) {
    if (localHost == null) {
      return new InetSocketAddress(localPort);
    }
    return new InetSocketAddress(localHost, localPort);
  }

}

    try {
      InetSocketAddress inetSocketAddress = createNewInetSocketAddress(localHost, localPort);
      final ServerSocketChannel server = ServerSocketChannel.open().bind(inetSocketAddress);

      final AtomicBoolean alive = new AtomicBoolean(true);
      final CopyOnWriteArrayList<PortForward> handles = new CopyOnWriteArrayList<>();

      final ExecutorService executorService = Executors.newSingleThreadExecutor();

      // Create a handle that can be used to retrieve information and stop the port-forward
      final LocalPortForward localPortForwardHandle = new LocalPortForward() {
        @Override
        public void close() throws IOException {
          alive.set(false);
          try {
            server.close();
          } finally {
            Utils.closeQuietly(handles);
            executorService.shutdownNow();
          }
        }

        @Override
        public boolean isAlive() {
          return alive.get();
        }

        @Override
        public boolean errorOccurred() {
          for (PortForward handle : handles) {
            if (handle.errorOccurred()) {
              return true;
            }
          }
          return false;
        }

        @Override
        public InetAddress getLocalAddress() {
          try {
            return ((InetSocketAddress) server.getLocalAddress()).getAddress();
          } catch (IOException e) {
            throw new IllegalStateException("Cannot determine local address", e);
          }
        }

        @Override
        public int getLocalPort() {
          try {
            return ((InetSocketAddress) server.getLocalAddress()).getPort();
          } catch (IOException e) {
            throw new IllegalStateException("Cannot determine local address", e);
          }
        }

        @Override
        public Collection<Throwable> getClientThrowables() {
          Collection<Throwable> clientThrowables = new ArrayList<>();
          for (PortForward handle : handles) {
            clientThrowables.addAll(handle.getClientThrowables());
          }
          return clientThrowables;
        }

        @Override
        public Collection<Throwable> getServerThrowables() {
          Collection<Throwable> serverThrowables = new ArrayList<>();
          for (PortForward handle : handles) {
            serverThrowables.addAll(handle.getServerThrowables());
          }
          return serverThrowables;
        }
      };

      // Start listening on localhost for new connections.
      // Every new connection will open its own stream on the remote resource.
      executorService.execute(() -> {
        // accept cycle
        while (alive.get()) {
          try {
            SocketChannel socket = server.accept();
            handles.add(forward(resourceBaseUrl, port, socket, socket));
          } catch (IOException e) {
            if (alive.get()) {
              LOG.error("Error while listening for connections", e);
            }
            Utils.closeQuietly(localPortForwardHandle);
          }
        }
      });

      return localPortForwardHandle;
    } catch (IOException e) {
      throw new IllegalStateException("Unable to port forward", e);
    }
  
<no_super_class>