/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.cache.client.internal;

import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.ServerConnectivityException;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.internal.AvailablePort;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.test.junit.categories.ClientServerTest;
import org.apache.geode.test.junit.categories.IntegrationTest;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.util.Properties;

import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.junit.Assert.*;

/**
 *
 */
@Category({IntegrationTest.class, ClientServerTest.class})
public class ConnectionPoolImplJUnitTest {

  private static final String expectedRedundantErrorMsg =
      "Could not find any server to create redundant client queue on.";
  private static final String expectedPrimaryErrorMsg =
      "Could not find any server to create primary client queue on.";

  private Cache cache;
  private int port;

  @Before
  public void setUp() {
    Properties props = new Properties();
    props.setProperty(MCAST_PORT, "0");
    props.setProperty(LOCATORS, "");
    cache = CacheFactory.create(DistributedSystem.connect(props));
    port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
  }

  @After
  public void tearDown() {
    if (cache != null && !cache.isClosed()) {
      unsetQueueError();
      cache.close();
    }
  }

  private void setQueueError() {
    final String addExpectedPEM =
        "<ExpectedException action=add>" + expectedPrimaryErrorMsg + "</ExpectedException>";
    final String addExpectedREM =
        "<ExpectedException action=add>" + expectedRedundantErrorMsg + "</ExpectedException>";

    cache.getLogger().info(addExpectedPEM);
    cache.getLogger().info(addExpectedREM);

  }

  private void unsetQueueError() {
    final String removeExpectedPEM =
        "<ExpectedException action=remove>" + expectedPrimaryErrorMsg + "</ExpectedException>";
    final String removeExpectedREM =
        "<ExpectedException action=remove>" + expectedRedundantErrorMsg + "</ExpectedException>";

    cache.getLogger().info(removeExpectedPEM);
    cache.getLogger().info(removeExpectedREM);
  }

  @Test
  public void testDefaults() throws Exception {
    setQueueError();
    PoolFactory cpf = PoolManager.createFactory();
    cpf.addServer("localhost", port);

    PoolImpl pool = (PoolImpl) cpf.create("myfriendlypool");

    // check defaults
    assertEquals(PoolFactory.DEFAULT_SOCKET_CONNECT_TIMEOUT, pool.getSocketConnectTimeout());
    assertEquals(PoolFactory.DEFAULT_FREE_CONNECTION_TIMEOUT, pool.getFreeConnectionTimeout());
    assertEquals(PoolFactory.DEFAULT_SOCKET_BUFFER_SIZE, pool.getSocketBufferSize());
    assertEquals(PoolFactory.DEFAULT_READ_TIMEOUT, pool.getReadTimeout());
    assertEquals(PoolFactory.DEFAULT_MIN_CONNECTIONS, pool.getMinConnections());
    assertEquals(PoolFactory.DEFAULT_MAX_CONNECTIONS, pool.getMaxConnections());
    assertEquals(PoolFactory.DEFAULT_RETRY_ATTEMPTS, pool.getRetryAttempts());
    assertEquals(PoolFactory.DEFAULT_IDLE_TIMEOUT, pool.getIdleTimeout());
    assertEquals(PoolFactory.DEFAULT_PING_INTERVAL, pool.getPingInterval());
    assertEquals(PoolFactory.DEFAULT_THREAD_LOCAL_CONNECTIONS, pool.getThreadLocalConnections());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_ENABLED, pool.getSubscriptionEnabled());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_REDUNDANCY, pool.getSubscriptionRedundancy());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_MESSAGE_TRACKING_TIMEOUT,
        pool.getSubscriptionMessageTrackingTimeout());
    assertEquals(PoolFactory.DEFAULT_SUBSCRIPTION_ACK_INTERVAL, pool.getSubscriptionAckInterval());
    assertEquals(PoolFactory.DEFAULT_SERVER_GROUP, pool.getServerGroup());
    // check non default
    assertEquals("myfriendlypool", pool.getName());
    assertEquals(1, pool.getServers().size());
    assertEquals(0, pool.getLocators().size());
    {
      InetSocketAddress addr = (InetSocketAddress) pool.getServers().get(0);
      assertEquals(port, addr.getPort());
      assertEquals("localhost", addr.getHostName());
    }

  }

  @Test
  public void testProperties() throws Exception {
    int readTimeout = 234234;
    int socketTimeout = 123123;

    PoolFactory cpf = PoolManager.createFactory();
    cpf.addServer("localhost", port).setSocketConnectTimeout(socketTimeout)
        .setReadTimeout(readTimeout).setThreadLocalConnections(true);

    PoolImpl pool = (PoolImpl) cpf.create("myfriendlypool");

    // check non default
    assertEquals("myfriendlypool", pool.getName());
    assertEquals(socketTimeout, pool.getSocketConnectTimeout());
    assertEquals(readTimeout, pool.getReadTimeout());
    assertEquals(true, pool.getThreadLocalConnections());
    assertEquals(1, pool.getServers().size());
    assertEquals(0, pool.getLocators().size());
    {
      InetSocketAddress addr = (InetSocketAddress) pool.getServers().get(0);
      assertEquals(port, addr.getPort());
      assertEquals("localhost", addr.getHostName());
    }
    LiveServerPinger lsp = new LiveServerPinger(pool);
    long NANOS_PER_MS = 1000000L;
    assertEquals(((pool.getPingInterval() + 1) / 2) * NANOS_PER_MS, lsp.pingIntervalNanos);
  }

  @Test
  public void testCacheClose() throws Exception {
    PoolFactory cpf = PoolManager.createFactory();
    cpf.addLocator("localhost", AvailablePortHelper.getRandomAvailableTCPPort());
    Pool pool1 = cpf.create("pool1");
    Pool pool2 = cpf.create("pool2");
    cache.close();

    assertTrue(pool1.isDestroyed());
    assertTrue(pool2.isDestroyed());
  }

  @Test
  public void testExecuteOp() throws Exception {
    CacheServer server1 = cache.addCacheServer();
    CacheServer server2 = cache.addCacheServer();
    int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(2);
    int port1 = ports[0];
    int port2 = ports[1];
    server1.setPort(port1);
    server2.setPort(port2);

    server1.start();
    server2.start();

    PoolFactory cpf = PoolManager.createFactory();
    cpf.addServer("localhost", port2);
    cpf.addServer("localhost", port1);
    PoolImpl pool = (PoolImpl) cpf.create("pool1");

    ServerLocation location1 = new ServerLocation("localhost", port1);
    ServerLocation location2 = new ServerLocation("localhost", port2);

    Op testOp = new Op() {
      int attempts = 0;

      public Object attempt(Connection cnx) throws Exception {
        if (attempts == 0) {
          attempts++;
          throw new SocketTimeoutException();
        } else {
          return cnx.getServer();
        }

      }

      @Override
      public boolean useThreadLocalConnection() {
        return true;
      }
    };

    // TODO - set retry attempts, and throw in some assertions
    // about how many times we retry

    ServerLocation usedServer = (ServerLocation) pool.execute(testOp);
    assertTrue("expected " + location1 + " or " + location2 + ", got " + usedServer,
        location1.equals(usedServer) || location2.equals(usedServer));

    testOp = new Op() {
      public Object attempt(Connection cnx) throws Exception {
        throw new SocketTimeoutException();
      }

      @Override
      public boolean useThreadLocalConnection() {
        return true;
      }
    };

    try {
      usedServer = (ServerLocation) pool.execute(testOp);
      fail("Should have failed");
    } catch (ServerConnectivityException expected) {
      // do nothing
    }
  }

  @Test
  public void testCreatePool() throws Exception {
    CacheServer server1 = cache.addCacheServer();
    int port1 = port;
    server1.setPort(port1);

    server1.start();

    PoolFactory cpf = PoolManager.createFactory();
    cpf.addServer("localhost", port1);
    cpf.setSubscriptionEnabled(true);
    cpf.setSubscriptionRedundancy(0);
    PoolImpl pool = (PoolImpl) cpf.create("pool1");

    ServerLocation location1 = new ServerLocation("localhost", port1);

    Op testOp = new Op() {
      public Object attempt(Connection cnx) throws Exception {
        return cnx.getServer();
      }

      @Override
      public boolean useThreadLocalConnection() {
        return true;
      }
    };

    assertEquals(location1, pool.executeOnPrimary(testOp));
    assertEquals(location1, pool.executeOnQueuesAndReturnPrimaryResult(testOp));
  }

}
