clicache/integration-test/ThinClientDurableTestsN.cs

/*
 * 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.
 */

using System;
using System.Collections.Generic;
using System.Threading;

namespace Apache.Geode.Client.UnitTests
{
  using NUnit.Framework;
  using Apache.Geode.DUnitFramework;
  using Apache.Geode.Client;


  using AssertionException = Apache.Geode.Client.AssertionException;
  
  [Ignore("broken")]
  [TestFixture]
  [Category("group2")]
  [Category("unicast_only")]
  [Category("generics")]
  public class ThinClientDurableTests : ThinClientRegionSteps
  {
    #region Private members

    private UnitProcess m_client1, m_client2, m_feeder;
    private string[] m_regexes = { "D-Key-.*", "Key-.*" };
    private string[] m_mixKeys = { "Key-1", "D-Key-1", "L-Key", "LD-Key" };
    private string[] keys = { "Key-1", "Key-2", "Key-3", "Key-4", "Key-5" };

    private static string DurableClientId1 = "DurableClientId1";
    private static string DurableClientId2 = "DurableClientId2";

    private static DurableListener<object, object> m_checker1, m_checker2;

    #endregion

    protected override ClientBase[] GetClients()
    {
      m_client1 = new UnitProcess();
      m_client2 = new UnitProcess();
      m_feeder = new UnitProcess();
      return new ClientBase[] { m_client1, m_client2, m_feeder };
    }

    [TestFixtureTearDown]
    public override void EndTests()
    {
      CacheHelper.StopJavaServers();
      base.EndTests();
    }

    [TearDown]
    public override void EndTest()
    {
      try
      {
        m_client1.Call(CacheHelper.Close);
        m_client2.Call(CacheHelper.Close);
        m_feeder.Call(CacheHelper.Close);
        CacheHelper.ClearEndpoints();
        CacheHelper.ClearLocators();
      }
      finally
      {
        CacheHelper.StopJavaServers();
      }
      base.EndTest();
    }

    #region Common Functions

    public void InitFeeder(string locators, int redundancyLevel)
    {
      CacheHelper.CreatePool<object, object>("__TESTPOOL1_", locators, (string)null, redundancyLevel, false);
      CacheHelper.CreateTCRegion_Pool<object, object>(RegionNames[0], false, true, null,
        locators, "__TESTPOOL1_", false);
    }

    public void InitFeeder2(string locators, int redundancyLevel)
    {
      CacheHelper.CreatePool<object, object>("__TESTPOOL1_", locators, (string)null, redundancyLevel, false);
      CacheHelper.CreateTCRegion_Pool<object, object>(RegionNames[0], false, true, null,
        locators, "__TESTPOOL1_", false);

      CacheHelper.CreatePool<object, object>("__TESTPOOL2_", locators, (string)null, redundancyLevel, false);
      CacheHelper.CreateTCRegion_Pool<object, object>(RegionNames[1], false, true, null,
        locators, "__TESTPOOL2_", false);
    }

    public void InitDurableClientWithTwoPools(string locators,
    int redundancyLevel, string durableClientId, TimeSpan durableTimeout, int expectedQ0, int expectedQ1)
    {
      DurableListener<object, object> checker = null;
      CacheHelper.InitConfigForDurable_Pool2(locators, redundancyLevel,
          durableClientId, durableTimeout, TimeSpan.FromSeconds(35), "__TESTPOOL1_");
      CacheHelper.CreateTCRegion_Pool(RegionNames[0], false, true, checker,
          CacheHelper.Locators, "__TESTPOOL1_", true);

      CacheHelper.InitConfigForDurable_Pool2(locators, redundancyLevel,
          durableClientId, durableTimeout, TimeSpan.FromSeconds(35), "__TESTPOOL2_");
      CacheHelper.CreateTCRegion_Pool(RegionNames[1], false, true, checker,
          CacheHelper.Locators, "__TESTPOOL2_", true);

      IRegion<object, object> region0 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[1]);

      try
      {
        region0.GetSubscriptionService().RegisterAllKeys(true);
        region1.GetSubscriptionService().RegisterAllKeys(true);
      }
      catch (Exception other)
      {
        Assert.Fail("RegisterAllKeys threw unexpected exception: {0}", other.Message);
      }

      Pool pool0 = CacheHelper.DCache.GetPoolManager().Find(region0.Attributes.PoolName);
      int pendingEventCount0 = pool0.PendingEventCount;
      Util.Log("pendingEventCount0 for pool = {0} {1} ", pendingEventCount0, region0.Attributes.PoolName);
      string msg = string.Format("Expected Value ={0}, Actual = {1}", expectedQ0, pendingEventCount0);
      Assert.AreEqual(expectedQ0, pendingEventCount0, msg);

      Pool pool1 = CacheHelper.DCache.GetPoolManager().Find(region1.Attributes.PoolName);
      int pendingEventCount1 = pool1.PendingEventCount;
      Util.Log("pendingEventCount1 for pool = {0} {1} ", pendingEventCount1, region1.Attributes.PoolName);
      string msg1 = string.Format("Expected Value ={0}, Actual = {1}", expectedQ1, pendingEventCount1);
      Assert.AreEqual(expectedQ1, pendingEventCount1, msg1);

      CacheHelper.DCache.ReadyForEvents();
      Thread.Sleep(10000);

      CacheHelper.DCache.Close(true);
    }

    public void ClearChecker(int client)
    {
      if (client == 1)
      {
        ThinClientDurableTests.m_checker1 = null;
      }
      else // client == 2
      {
        ThinClientDurableTests.m_checker2 = null;
      }
    }

    public void InitDurableClient(int client, string locators, int redundancyLevel,
      string durableClientId, TimeSpan durableTimeout)
    {
      // Create DurableListener for first time and use same afterward.
      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        if (ThinClientDurableTests.m_checker1 == null)
        {
          ThinClientDurableTests.m_checker1 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2 
      {
        if (ThinClientDurableTests.m_checker2 == null)
        {
          ThinClientDurableTests.m_checker2 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker2;
      }
      CacheHelper.InitConfigForDurable_Pool(locators, redundancyLevel,
        durableClientId, durableTimeout);
      CacheHelper.CreateTCRegion_Pool<object, object>(RegionNames[0], false, true, checker,
        CacheHelper.Locators, "__TESTPOOL1_", true);

      CacheHelper.DCache.ReadyForEvents();
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);
      region1.GetSubscriptionService().RegisterRegex(m_regexes[0], true);
      region1.GetSubscriptionService().RegisterRegex(m_regexes[1], false);
      //CacheableKey[] ldkeys = { new CacheableString(m_mixKeys[3]) };
      ICollection<object> lkeys = new List<object>();
      lkeys.Add((object)m_mixKeys[3]);
      region1.GetSubscriptionService().RegisterKeys(lkeys, true, false);

      ICollection<object> ldkeys = new List<object>(); ;
      ldkeys.Add((object)m_mixKeys[2]);
      region1.GetSubscriptionService().RegisterKeys(ldkeys, false, false);
    }

    public void InitClientXml(string cacheXml)
    {
      CacheHelper.InitConfig(cacheXml);
    }

    public void ReadyForEvents()
    {
      CacheHelper.DCache.ReadyForEvents();
    }

    public void PendingEventCount(IRegion<object, object> region, int expectedPendingQSize, bool exception)
    {
      Util.Log("PendingEventCount regionName = {0} ", region);
      string poolName = region.Attributes.PoolName;
      if (poolName != null)
      {
        Util.Log("PendingEventCount poolName = {0} ", poolName);
        Pool pool = CacheHelper.DCache.GetPoolManager().Find(poolName);
        if (exception)
        {
          try
          {
            int pendingEventCount = pool.PendingEventCount;
            Util.Log("PendingEventCount Should have got exception ");
            Assert.Fail("PendingEventCount Should have got exception");
          }
          catch (IllegalStateException ex)
          {
            Util.Log("Got expected exception for PendingEventCount {0} ", ex.Message);
          }
        }
        else
        {
          int pendingEventCount = pool.PendingEventCount;
          Util.Log("pendingEventCount = {0} ", pendingEventCount);
          string msg = string.Format("Expected Value ={0}, Actual = {1}", expectedPendingQSize, pendingEventCount);
          Assert.AreEqual(expectedPendingQSize, pendingEventCount, msg);
        }
      }
    }

    public void FeederUpdate(int value, int sleep)
    {
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);

      region1[m_mixKeys[0]] = value;
      Thread.Sleep(sleep);
      region1[m_mixKeys[1]] = value;
      Thread.Sleep(sleep);
      region1[m_mixKeys[2]] = value;
      Thread.Sleep(sleep);
      region1[m_mixKeys[3]] = value;
      Thread.Sleep(sleep);

      region1.Remove(m_mixKeys[0]);
      Thread.Sleep(sleep);
      region1.Remove(m_mixKeys[1]);
      Thread.Sleep(sleep);
      region1.Remove(m_mixKeys[2]);
      Thread.Sleep(sleep);
      region1.Remove(m_mixKeys[3]);
      Thread.Sleep(sleep);
    }

    public void FeederUpdate2(int pool1, int pool2)
    {
      IRegion<object, object> region0 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[1]);

      for (int i = 0; i < pool1; i++)
      {
        region0[i] = i;
      }

      for (int i = 0; i < pool2; i++)
      {
        region1[i] = i;
      }
    }

    public void ClientDown(bool keepalive)
    {
      if (keepalive)
      {
        CacheHelper.CloseKeepAlive();
      }
      else
      {
        CacheHelper.Close();
      }
    }

    public void CrashClient()
    {
      // TODO:  crash client here.
    }

    public void KillServer()
    {
      CacheHelper.StopJavaServer(1);
      Util.Log("Cacheserver 1 stopped.");
    }

    public delegate void KillServerDelegate();

    #endregion


    public void VerifyTotal(int client, int keys, int total)
    {
      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2
      {
        checker = ThinClientDurableTests.m_checker2;
      }

      if (checker != null)
      {
        checker.validate(keys, total);
      }
      else
      {
        Assert.Fail("Checker is NULL!");
      }
    }

    public void VerifyBasic(int client, int keyCount, int eventCount, int durableValue, int nonDurableValue)
    {//1 4 8 1 1 
      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2
      {
        checker = ThinClientDurableTests.m_checker2;
      }

      if (checker != null)
      {
        try
        {
          checker.validateBasic(keyCount, eventCount, durableValue, nonDurableValue);//4 8 1 1 
        }
        catch (AssertionException e)
        {
          Util.Log("VERIFICATION FAILED for client {0}: {1} ", client, e);
          throw e;
        }
      }
      else
      {
        Assert.Fail("Checker is NULL!");
      }
    }

    #region Basic Durable Test


    void runDurableAndNonDurableBasic()
    {
      CacheHelper.SetupJavaServers(true,
        "cacheserver_notify_subscription.xml", "cacheserver_notify_subscription2.xml");
      CacheHelper.StartJavaLocator(1, "GFELOC");

      for (int redundancy = 0; redundancy <= 1; redundancy++)
      {
        for (int closeType = 1; closeType <= 2; closeType++)
        {
          for (int downtime = 0; downtime <= 1; downtime++) // downtime updates
          {
            Util.Log("Starting loop with closeType = {0}, redundancy = {1}, downtime = {2} ", closeType, redundancy, downtime);

            CacheHelper.StartJavaServerWithLocators(1, "GFECS1", 1);
            Util.Log("Cacheserver 1 started.");

            if (redundancy == 1)
            {
              CacheHelper.StartJavaServerWithLocators(2, "GFECS2", 1);
              Util.Log("Cacheserver 2 started.");
            }

            m_feeder.Call(InitFeeder, CacheHelper.Locators, 0);
            Util.Log("Feeder initialized.");

            m_client1.Call(ClearChecker, 1);
            m_client2.Call(ClearChecker, 2);

            m_client1.Call(InitDurableClient, 1, CacheHelper.Locators, redundancy, DurableClientId1, TimeSpan.FromSeconds(300));
            m_client2.Call(InitDurableClient, 2, CacheHelper.Locators, redundancy, DurableClientId2, TimeSpan.FromSeconds(3));

            Util.Log("Clients initialized.");

            m_feeder.Call(FeederUpdate, 1, 10);

            Util.Log("Feeder performed first update.");
            Thread.Sleep(45000); // wait for HA Q to drain and notify ack to go out.

            switch (closeType)
            {
              case 1:

                m_client1.Call(ClientDown, true);
                m_client2.Call(ClientDown, true);

                Util.Log("Clients downed with keepalive true.");
                break;
              case 2:

                m_client1.Call(ClientDown, false);
                m_client2.Call(ClientDown, false);

                Util.Log("Clients downed with keepalive false.");
                break;
              case 3:

                m_client1.Call(CrashClient);

                m_client2.Call(CrashClient);

                Util.Log("Clients downed as crash.");
                break;
              default:
                break;
            }

            if (downtime == 1)
            {
              m_feeder.Call(FeederUpdate, 2, 10);

              Util.Log("Feeder performed update during downtime.");
              Thread.Sleep(20000); // wait for HA Q to drain and notify ack to go out.
            }

            m_client1.Call(InitDurableClient, 1, CacheHelper.Locators, redundancy, DurableClientId1, TimeSpan.FromSeconds(300));

            // Sleep for 45 seconds since durable timeout is 30 seconds so that client2 times out
            Thread.Sleep(45000);

            m_client2.Call(InitDurableClient, 2, CacheHelper.Locators, redundancy, DurableClientId2, TimeSpan.FromSeconds(30));

            Util.Log("Clients brought back up.");

            if (closeType != 2 && downtime == 1)
            {
              m_client1.Call(VerifyBasic, 1, 4, 12, 2, 1);

              m_client2.Call(VerifyBasic, 2, 4, 8, 1, 1);

            }
            else
            {

              m_client1.Call(VerifyBasic, 1, 4, 8, 1, 1);

              m_client2.Call(VerifyBasic, 2, 4, 8, 1, 1);

            }

            Util.Log("Verification completed.");

            m_feeder.Call(ClientDown, false);

            m_client1.Call(ClientDown, false);

            m_client2.Call(ClientDown, false);

            Util.Log("Feeder and Clients closed.");

            CacheHelper.StopJavaServer(1);
            Util.Log("Cacheserver 1 stopped.");

            if (redundancy == 1)
            {
              CacheHelper.StopJavaServer(2);
              Util.Log("Cacheserver 2 stopped.");
            }

            Util.Log("Completed loop with closeType = {0}, redundancy = {1}, downtime = {2} ", closeType, redundancy, downtime);

          } // end for int downtime
        } // end for int closeType
      } // end for int redundancy
      CacheHelper.StopJavaLocator(1);
    }

    // Basic Durable Test to check durable event recieving for different combination
    // of Close type ( Keep Alive = true / false ) , Intermediate update and rudundancy

    [Test]
    public void DurableAndNonDurableBasic()
    {
      runDurableAndNonDurableBasic();
    } // end [Test] DurableAndNonDurableBasic

    #endregion

    #region Durable Intrest Test

    public void InitDurableClientRemoveInterest(int client, string locators,
      int redundancyLevel, string durableClientId, TimeSpan durableTimeout)
    {
      // Client Registered Durable Intrest on two keys. We need to unregister them all here.

      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        if (ThinClientDurableTests.m_checker1 == null)
        {
          ThinClientDurableTests.m_checker1 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2
      {
        if (ThinClientDurableTests.m_checker2 == null)
        {
          ThinClientDurableTests.m_checker2 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker2;
      }
      CacheHelper.InitConfigForDurable_Pool(locators, redundancyLevel,
        durableClientId, durableTimeout);
      CacheHelper.CreateTCRegion_Pool(RegionNames[0], false, true, checker,
        CacheHelper.Locators, "__TESTPOOL1_", true);

      CacheHelper.DCache.ReadyForEvents();
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);

      // Unregister Regex only durable
      region1.GetSubscriptionService().RegisterRegex(m_regexes[0], true);
      region1.GetSubscriptionService().UnregisterRegex(m_regexes[0]);

      // Unregister list only durable
      string[] ldkeys = new string[] { m_mixKeys[3] };
      region1.GetSubscriptionService().RegisterKeys(ldkeys, true, false);
      region1.GetSubscriptionService().UnregisterKeys(ldkeys);
    }

    public void InitDurableClientNoInterest(int client, string locators,
      int redundancyLevel, string durableClientId, TimeSpan durableTimeout)
    {
      // we use "client" to either create a DurableListener or use the existing ones
      // if the clients are initialized for the second time
      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        if (ThinClientDurableTests.m_checker1 == null)
        {
          ThinClientDurableTests.m_checker1 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2
      {
        if (ThinClientDurableTests.m_checker2 == null)
        {
          ThinClientDurableTests.m_checker2 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker2;
      }
      CacheHelper.InitConfigForDurable_Pool(locators, redundancyLevel,
        durableClientId, durableTimeout);
      CacheHelper.CreateTCRegion_Pool(RegionNames[0], false, true, checker,
        CacheHelper.Locators, "__TESTPOOL1_", true);
      CacheHelper.DCache.ReadyForEvents();
    }

    void runDurableInterest()
    {
      CacheHelper.SetupJavaServers(true, "cacheserver_notify_subscription.xml");
      CacheHelper.StartJavaLocator(1, "GFELOC");
      Util.Log("Locator started");
      CacheHelper.StartJavaServerWithLocators(1, "GFECS1", 1);
      Util.Log("Cacheserver 1 started.");

      m_feeder.Call(InitFeeder, CacheHelper.Locators, 0);
      Util.Log("Feeder started.");

      m_client1.Call(ClearChecker, 1);
      m_client2.Call(ClearChecker, 2);
      m_client1.Call(InitDurableClient, 1, CacheHelper.Locators,
        0, DurableClientId1, TimeSpan.FromSeconds(60));
      m_client2.Call(InitDurableClient, 2, CacheHelper.Locators,
        0, DurableClientId2, TimeSpan.FromSeconds(60));
      Util.Log("Clients started.");

      m_feeder.Call(FeederUpdate, 1, 10);
      Util.Log("Feeder performed first update.");

      Thread.Sleep(15000);

      m_client1.Call(ClientDown, true);
      m_client2.Call(ClientDown, true);
      Util.Log("Clients downed with keepalive true.");

      m_client1.Call(InitDurableClientNoInterest, 1, CacheHelper.Locators,
        0, DurableClientId1, TimeSpan.FromSeconds(60));
      Util.Log("Client 1 started with no interest.");

      m_client2.Call(InitDurableClientRemoveInterest, 2, CacheHelper.Locators,
        0, DurableClientId2, TimeSpan.FromSeconds(60));
      Util.Log("Client 2 started with remove interest.");

      m_feeder.Call(FeederUpdate, 2, 10);
      Util.Log("Feeder performed second update.");

      Thread.Sleep(10000);

      // only durable Intrest will remain.
      m_client1.Call(VerifyBasic, 1, 4, 12, 2, 1);

      // no second update should be recieved.
      m_client2.Call(VerifyBasic, 2, 4, 8, 1, 1);
      Util.Log("Verification completed.");

      m_feeder.Call(ClientDown, false);
      m_client1.Call(ClientDown, false);
      m_client2.Call(ClientDown, false);
      Util.Log("Feeder and Clients closed.");

      CacheHelper.StopJavaServer(1);
      Util.Log("Cacheserver 1 stopped.");

      CacheHelper.StopJavaLocator(1);
      Util.Log("Locator stopped");

      CacheHelper.ClearEndpoints();
      CacheHelper.ClearLocators();
    }

    //This is to test whether durable registered intrests remains on reconnect. and 
    // Unregister works on reconnect.

    [Test]
    public void DurableInterest()
    {
      runDurableInterest();
    } // end [Test] DurableInterest
    #endregion

    #region Durable Failover Test


    public void InitDurableClientForFailover(int client, string locators,
      int redundancyLevel, string durableClientId, TimeSpan durableTimeout)
    {
      // we use "client" to either create a DurableListener or use the existing ones
      // if the clients are initialized for the second time
      DurableListener<object, object> checker = null;
      if (client == 1)
      {
        if (ThinClientDurableTests.m_checker1 == null)
        {
          ThinClientDurableTests.m_checker1 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker1;
      }
      else // client == 2
      {
        if (ThinClientDurableTests.m_checker2 == null)
        {
          ThinClientDurableTests.m_checker2 = DurableListener<object, object>.Create();
        }
        checker = ThinClientDurableTests.m_checker2;
      }
      CacheHelper.InitConfigForDurable_Pool(locators, redundancyLevel,
        durableClientId, durableTimeout, TimeSpan.FromSeconds(35));
      CacheHelper.CreateTCRegion_Pool(RegionNames[0], false, true, checker,
        CacheHelper.Locators, "__TESTPOOL1_", true);
      CacheHelper.DCache.ReadyForEvents();
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(RegionNames[0]);

      try
      {
        region1.GetSubscriptionService().RegisterRegex(m_regexes[0], true);
        region1.GetSubscriptionService().RegisterRegex(m_regexes[1], false);
      }
      catch (Exception other)
      {
        Assert.Fail("RegisterKeys threw unexpected exception: {0}", other.Message);
      }
    }

    public void FeederUpdateForFailover(string region, int value, int sleep)
    {
      //update only 2 keys.
      IRegion<object, object> region1 = CacheHelper.GetVerifyRegion<object, object>(region);

      region1[m_mixKeys[0]] = value;
      Thread.Sleep(sleep);
      region1[m_mixKeys[1]] = value;
      Thread.Sleep(sleep);

    }

    void runDurableFailover()
    {
      CacheHelper.SetupJavaServers(true,
        "cacheserver_notify_subscription.xml", "cacheserver_notify_subscription2.xml");

      CacheHelper.StartJavaLocator(1, "GFELOC");
      Util.Log("Locator started");

      for (int clientDown = 0; clientDown <= 1; clientDown++)
      {
        for (int redundancy = 0; redundancy <= 1; redundancy++)
        {
          Util.Log("Starting loop with clientDown = {0}, redundancy = {1}", clientDown, redundancy);

          CacheHelper.StartJavaServerWithLocators(1, "GFECS1", 1);
          Util.Log("Cacheserver 1 started.");

          m_feeder.Call(InitFeeder, CacheHelper.Locators, 0);
          Util.Log("Feeder started with redundancy level as 0.");

          m_client1.Call(ClearChecker, 1);
          m_client1.Call(InitDurableClientForFailover, 1, CacheHelper.Locators,
            redundancy, DurableClientId1, TimeSpan.FromSeconds(300));
          Util.Log("Client started with redundancy level as {0}.", redundancy);

          m_feeder.Call(FeederUpdateForFailover, RegionNames[0], 1, 10);
          Util.Log("Feeder updates 1 completed.");

          CacheHelper.StartJavaServerWithLocators(2, "GFECS2", 1);
          Util.Log("Cacheserver 2 started.");

          //Time for redundancy thread to detect.
          Thread.Sleep(35000);

          if (clientDown == 1)
          {
            m_client1.Call(ClientDown, true);
          }

          CacheHelper.StopJavaServer(1);
          Util.Log("Cacheserver 1 stopped.");

          //Time for failover
          Thread.Sleep(5000);

          m_feeder.Call(FeederUpdateForFailover, RegionNames[0], 2, 10);
          Util.Log("Feeder updates 2 completed.");

          //Restart Client
          if (clientDown == 1)
          {
            m_client1.Call(InitDurableClientForFailover, 1, CacheHelper.Locators,
              redundancy, DurableClientId1, TimeSpan.FromSeconds(300));
            Util.Log("Client Restarted with redundancy level as {0}.", redundancy);
          }

          //Verify
          if (clientDown == 1)
          {
            if (redundancy == 0) // Events missed
            {
              m_client1.Call(VerifyBasic, 1, 2, 2, 1, 1);
            }
            else // redundancy == 1 Only Durable Events should be recieved.
            {
              m_client1.Call(VerifyBasic, 1, 2, 3, 2, 1);
            }
          }
          else  // In normal failover all events should be recieved.
          {
            m_client1.Call(VerifyBasic, 1, 2, 4, 2, 2);
          }

          Util.Log("Verification completed.");

          m_feeder.Call(ClientDown, false);
          m_client1.Call(ClientDown, false);
          Util.Log("Feeder and Client closed.");

          CacheHelper.StopJavaServer(2);
          Util.Log("Cacheserver 2 stopped.");

          Util.Log("Completed loop with clientDown = {0}, redundancy = {1}", clientDown, redundancy);
        }// for redundancy
      } // for clientDown
      CacheHelper.StopJavaLocator(1);
      Util.Log("Locator stopped");

      CacheHelper.ClearEndpoints();
      CacheHelper.ClearLocators();
    }

    void RunDurableClient(int expectedPendingQSize)
    {
      var pp = Properties<string, string>.Create();
      pp.Insert("durable-client-id", "DurableClientId");
      pp.Insert("durable-timeout", "30s");

      var cacheFactory = new CacheFactory(pp);
      var cache = cacheFactory.Create();
      cache.GetPoolFactory().SetSubscriptionEnabled(true);
      cache.GetPoolFactory().SetSubscriptionAckInterval(TimeSpan.FromMilliseconds(5000));
      cache.GetPoolFactory().SetSubscriptionMessageTrackingTimeout(TimeSpan.FromMilliseconds(5000));
      Util.Log("Created the Geode Cache Programmatically");

      RegionFactory regionFactory = cache.CreateRegionFactory(RegionShortcut.CACHING_PROXY);
      IRegion<object, object> region = regionFactory.Create<object, object>("DistRegionAck");
      Util.Log("Created the DistRegionAck Region Programmatically");

      QueryService qService = cache.GetQueryService();
      CqAttributesFactory<object, object> cqFac = new CqAttributesFactory<object, object>();

      ICqListener<object, object> cqLstner = new MyCqListener1<object, object>();
      cqFac.AddCqListener(cqLstner);
      CqAttributes<object, object> cqAttr = cqFac.Create();
      Util.Log("Attached CqListener");
      String query = "select * from /DistRegionAck";
      CqQuery<object, object> qry = qService.NewCq("MyCq", query, cqAttr, true);
      Util.Log("Created new CqQuery");

      qry.Execute();
      Util.Log("Executed new CqQuery");
      Thread.Sleep(10000);

      PendingEventCount(region, expectedPendingQSize, false);

      //Send ready for Event message to Server( only for Durable Clients ).
      //Server will send queued events to client after recieving this.
      cache.ReadyForEvents();

      Util.Log("Sent ReadyForEvents message to server");
      Thread.Sleep(10000);
      // Close the Geode Cache with keepalive = true.  Server will queue events for
      // durable registered keys and will deliver all events when client will reconnect
      // within timeout period and send "readyForEvents()"

      PendingEventCount(region, 0, true);

      cache.Close(true);

      Util.Log("Closed the Geode Cache with keepalive as true");
    }

    void runDurableClientWithTwoPools()
    {
      CacheHelper.SetupJavaServers(true, "cacheserver_notify_subscription.xml");
      CacheHelper.StartJavaLocator(1, "GFELOC");
      Util.Log("Locator started");
      CacheHelper.StartJavaServerWithLocators(1, "GFECS1", 1);
      Util.Log("Cacheserver 1 started.");

      m_feeder.Call(InitFeeder2, CacheHelper.Locators, 0);
      Util.Log("Feeder started.");

      m_client1.Call(InitDurableClientWithTwoPools, CacheHelper.Locators, 0, DurableClientId1, TimeSpan.FromSeconds(30), -2, -2);
      Util.Log("DurableClient with Two Pools Initialized");

      m_feeder.Call(FeederUpdate2, 5, 10);
      Util.Log("Feeder performed first update.");
      Thread.Sleep(15000);

      m_client1.Call(InitDurableClientWithTwoPools, CacheHelper.Locators, 0, DurableClientId1, TimeSpan.FromSeconds(30), 6, 11); //+1 for marker, so 5+1, 10+1 etc
      Util.Log("DurableClient with Two Pools after first update");

      m_feeder.Call(FeederUpdate2, 10, 5);
      Util.Log("Feeder performed second update.");
      Thread.Sleep(15000);

      m_client1.Call(InitDurableClientWithTwoPools, CacheHelper.Locators, 0, DurableClientId1, TimeSpan.FromSeconds(30), 16, 16);
      Util.Log("DurableClient with Two Pools after second update");

      Thread.Sleep(45000); //45 > 30 secs.
      m_client1.Call(InitDurableClientWithTwoPools, CacheHelper.Locators, 0, DurableClientId1, TimeSpan.FromSeconds(30), -1, -1);
      Util.Log("DurableClient with Two Pools after timeout");

      m_feeder.Call(ClientDown, false);
      Util.Log("Feeder and Clients closed.");

      CacheHelper.StopJavaServer(1);
      Util.Log("Cacheserver 1 stopped.");

      CacheHelper.StopJavaLocator(1);
      Util.Log("Locator stopped");

      CacheHelper.ClearEndpoints();
      CacheHelper.ClearLocators();
    }

    void RunFeeder()
    {
      var cacheFactory = new CacheFactory();
      Util.Log("Feeder connected to the Geode Distributed System");

      Cache cache = cacheFactory.Create();
      Util.Log("Created the Geode Cache");

      RegionFactory regionFactory = cache.CreateRegionFactory(RegionShortcut.PROXY);
      Util.Log("Created the RegionFactory");

      // Create the Region Programmatically.
      IRegion<object, object> region = regionFactory.Create<object, object>("DistRegionAck");
      Util.Log("Created the Region Programmatically.");

      PendingEventCount(region, 0, true);

      for (int i = 0; i < 10; i++)
      {
        region[i] = i;
      }
      Thread.Sleep(10000);
      Util.Log("put on 0-10 keys done.");

      // Close the Geode Cache
      cache.Close();
      Util.Log("Closed the Geode Cache");
    }

    void RunFeeder1()
    {
      CacheFactory cacheFactory = new CacheFactory();
      Util.Log("Feeder connected to the Geode Distributed System");

      Cache cache = cacheFactory.Create();
      Util.Log("Created the Geode Cache");

      RegionFactory regionFactory = cache.CreateRegionFactory(RegionShortcut.PROXY);
      Util.Log("Created the RegionFactory");

      // Create the Region Programmatically.
      IRegion<object, object> region = regionFactory.Create<object, object>("DistRegionAck");
      Util.Log("Created the Region Programmatically.");

      PendingEventCount(region, 0, true);

      for (int i = 10; i < 20; i++)
      {
        region[i] = i;
      }
      Thread.Sleep(10000);
      Util.Log("put on 10-20 keys done.");

      // Close the Geode Cache
      cache.Close();
      Util.Log("Closed the Geode Cache");
    }

    void VerifyEvents()
    {
      Util.Log("MyCqListener1.m_cntEvents = {0} ", MyCqListener1<object, object>.m_cntEvents);
      Assert.AreEqual(MyCqListener1<object, object>.m_cntEvents, 20, "Incorrect events, expected 20");
    }

    void runCQDurable()
    {
      CacheHelper.SetupJavaServers(false, "serverDurableClient.xml");
      CacheHelper.StartJavaServer(1, "GFECS1");
      m_client1.Call(RunDurableClient, -2); // 1st time no Q, hence check -2 as PendingEventCount.
      m_client2.Call(RunFeeder);
      m_client1.Call(RunDurableClient, 10);
      m_client2.Call(RunFeeder1);
      m_client1.Call(RunDurableClient, 10);
      m_client1.Call(VerifyEvents);
      Thread.Sleep(45 * 1000); // sleep 45 secs > 30 secs, check -1 as PendingEventCount.
      m_client1.Call(RunDurableClient, -1);
      CacheHelper.StopJavaServer(1);
    }

    [Test]
    public void DurableFailover()
    {
      runDurableFailover();
    } // end [Test] DurableFailover

    [Test]
    public void CQDurable()
    {
      runCQDurable();

      runDurableClientWithTwoPools();
    }
    #endregion
  }
}