/*
 * Copyright (c) 2011-2017 Contributors to the Eclipse Foundation
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
 * which is available at https://www.apache.org/licenses/LICENSE-2.0.
 *
 * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
 */

package io.vertx.core.http.impl;

import io.netty.channel.Channel;
import io.vertx.core.Handler;
import io.vertx.core.http.HttpConnection;
import io.vertx.core.http.HttpVersion;
import io.vertx.core.http.impl.pool.Pool;
import io.vertx.core.http.impl.pool.Waiter;
import io.vertx.core.impl.ContextInternal;
import io.vertx.core.spi.metrics.HttpClientMetrics;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;

/**
 * The connection manager associates remote hosts with pools, it also tracks all connections so they can be closed
 * when the manager is closed.
 *
 * @author <a href="http://tfox.org">Tim Fox</a>
 */
class ConnectionManager {

  private final int maxWaitQueueSize;
  private final HttpClientMetrics metrics; // Shall be removed later combining the PoolMetrics with HttpClientMetrics
  private final HttpClientImpl client;
  private final Map<Channel, HttpClientConnection> connectionMap = new ConcurrentHashMap<>();
  private final Map<EndpointKey, Endpoint> endpointMap = new ConcurrentHashMap<>();
  private final HttpVersion version;
  private final long maxSize;

  ConnectionManager(HttpClientImpl client,
                    HttpClientMetrics metrics,
                    HttpVersion version,
                    long maxSize,
                    int maxWaitQueueSize) {
    this.client = client;
    this.maxWaitQueueSize = maxWaitQueueSize;
    this.metrics = metrics;
    this.maxSize = maxSize;
    this.version = version;
  }

  private static final class EndpointKey {

    private final boolean ssl;
    private final int port;
    private final String peerHost;
    private final String host;

    EndpointKey(boolean ssl, int port, String peerHost, String host) {
      if (host == null) {
        throw new NullPointerException("No null host");
      }
      if (peerHost == null) {
        throw new NullPointerException("No null peer host");
      }
      this.ssl = ssl;
      this.peerHost = peerHost;
      this.host = host;
      this.port = port;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;
      EndpointKey that = (EndpointKey) o;
      return ssl == that.ssl && port == that.port && peerHost.equals(that.peerHost) && host.equals(that.host);
    }

    @Override
    public int hashCode() {
      int result = ssl ? 1 : 0;
      result = 31 * result + peerHost.hashCode();
      result = 31 * result + host.hashCode();
      result = 31 * result + port;
      return result;
    }
  }

  class Endpoint {

    private final Pool<HttpClientConnection> pool;
    private final Object metric;

    public Endpoint(Pool<HttpClientConnection> pool, Object metric) {
      this.pool = pool;
      this.metric = metric;
    }
  }

  void getConnection(String peerHost, boolean ssl, int port, String host,
                     Handler<HttpConnection> connectionHandler,
                     BiFunction<ContextInternal, HttpClientConnection, Boolean> onSuccess,
                     BiConsumer<ContextInternal, Throwable> onFailure) {
    EndpointKey key = new EndpointKey(ssl, port, peerHost, host);
    while (true) {
      Endpoint endpoint = endpointMap.computeIfAbsent(key, targetAddress -> {
        int maxPoolSize = Math.max(client.getOptions().getMaxPoolSize(), client.getOptions().getHttp2MaxPoolSize());
        Object metric = metrics != null ? metrics.createEndpoint(host, port, maxPoolSize) : null;
        HttpChannelConnector connector = new HttpChannelConnector(client, metric, version, ssl, peerHost, host, port);
        Pool<HttpClientConnection> pool = new Pool<>(connector, maxWaitQueueSize, maxSize,
          v -> {
            if (metrics != null) {
              metrics.closeEndpoint(host, port, metric);
            }
            endpointMap.remove(key);
          },
          connectionMap::put,
          connectionMap::remove);
        return new Endpoint(pool, metric);
      });
      Object metric;
      if (metrics != null) {
        metric = metrics.enqueueRequest(endpoint.metric);
      } else {
        metric = null;
      }
      if (endpoint.pool.getConnection(new Waiter<HttpClientConnection>(client.getVertx().getOrCreateContext()) {
        @Override
        public void initConnection(ContextInternal ctx, HttpClientConnection conn) {
          if (connectionHandler != null) {
            ctx.executeFromIO(() -> {
              connectionHandler.handle(conn);
            });
          }
        }
        @Override
        public void handleFailure(ContextInternal ctx, Throwable failure) {
          if (metrics != null) {
            metrics.dequeueRequest(endpoint.metric, metric);
          }
          onFailure.accept(ctx, failure);
        }
        @Override
        public boolean handleConnection(ContextInternal ctx, HttpClientConnection conn) throws Exception {
          if (metrics != null) {
            metrics.dequeueRequest(endpoint.metric, metric);
          }
          return onSuccess.apply(ctx, conn);
        }
      })) {
        break;
      }
    }
  }

  public void close() {
    endpointMap.clear();
    for (HttpClientConnection conn : connectionMap.values()) {
      conn.close();
    }
  }
}
