/*
 * 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.internal.tcp;

import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.apache.geode.test.dunit.Disconnect.disconnectAllFromDS;
import static org.apache.geode.test.dunit.VM.getController;
import static org.apache.geode.test.dunit.VM.getVM;
import static org.apache.geode.test.dunit.VM.toArray;
import static org.assertj.core.api.Assertions.assertThat;

import java.io.Serializable;

import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;

import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionImpl;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.rules.CacheRule;
import org.apache.geode.test.dunit.rules.DistributedRule;

public class CloseConnectionTest implements Serializable {
  private static final long serialVersionUID = 3692493564204797623L;
  private VM vm0;
  private VM vm1;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();

  /**
   * The test case in this class requires that conserve-sockets=true in order for the connections to
   * be shared.
   */
  @Rule
  public CacheRule cacheRule = new CacheRule.Builder().addConfig(CONSERVE_SOCKETS, "true").build();

  @Before
  public void setUp() {
    vm0 = getVM(0);
    vm1 = getVM(1);
  }

  @After
  public void tearDown() {
    for (VM vm : toArray(getController(), vm0, vm1)) {
      vm.invoke(() -> cacheRule.closeAndNullCache());
    }

    disconnectAllFromDS();
  }

  @Test
  public void sharedSenderShouldRecoverFromClosedSocket() {
    // Create a region in each member. VM0 has a proxy region, so state must be in VM1
    vm0.invoke(() -> {
      getCache().createRegionFactory(RegionShortcut.REPLICATE_PROXY).create("region");
    });
    vm1.invoke(() -> {
      getCache().createRegionFactory(RegionShortcut.REPLICATE).create("region");
    });


    // Force VM1 to close it's connections.
    vm1.invoke(() -> {
      ConnectionTable conTable = getConnectionTable();
      assertThat(conTable.getNumberOfReceivers()).isEqualTo(2);
      conTable.closeReceivers(false);
      await().untilAsserted(() -> assertThat(conTable.getNumberOfReceivers()).isEqualTo(0));
    });

    // See if VM0 noticed the closed connections. Try to do a couple of region
    // operations
    vm0.invoke(() -> {
      Region<Object, Object> region = getCache().getRegion("region");
      region.put("1", "1");

      assertThat(region.get("1")).isEqualTo("1");
    });

    // Make sure connections were reestablished
    vm1.invoke(() -> {
      ConnectionTable conTable = getConnectionTable();
      await().untilAsserted(() -> assertThat(conTable.getNumberOfReceivers()).isEqualTo(2));
    });

    // Make sure the Sender connection has a reader thread
    vm1.invoke(() -> {
      ConnectionTable conTable = getConnectionTable();
      InternalDistributedSystem distributedSystem = getCache().getInternalDistributedSystem();
      InternalDistributedMember otherMember = distributedSystem.getDistributionManager()
          .getOtherNormalDistributionManagerIds().iterator().next();
      Connection connection = conTable.getConduit().getConnection(otherMember, true,
          System.currentTimeMillis(), 15000, 0);
      await().untilAsserted(() -> {
        // grab the shared, ordered "sender" connection to vm0. It should have a residual
        // reader thread that exists to detect that the socket has been closed.
        ConnectionTable.threadWantsSharedResources();
        assertThat(connection.hasResidualReaderThread()).isTrue();
      });
    });
  }

  private ConnectionTable getConnectionTable() {
    ClusterDistributionManager cdm =
        (ClusterDistributionManager) getCache().getDistributionManager();
    DistributionImpl distribution = (DistributionImpl) cdm.getDistribution();
    return distribution.getDirectChannel().getConduit().getConTable();
  }

  private InternalCache getCache() {
    return cacheRule.getOrCreateCache();
  }
}