core/org.eclipse.smila.zookeeper.test/code/src/org/eclipse/smila/zookeeper/test/TestZkConfigurationUpdateWatcher.java

/*********************************************************************************************************************
 * Copyright (c) 2008, 2015 Empolis Information Management GmbH and brox IT Solutions GmbH. All rights reserved.
 * This program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 
 * which accompanies this distribution, and is available at http://www.eclipse.org/legal/epl-v10.html
**********************************************************************************************************************/
package org.eclipse.smila.zookeeper.test;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import org.eclipse.smila.test.DeclarativeServiceTestCase;
import org.eclipse.smila.utils.config.ConfigurationUpdateWatcher.UpdateableService;
import org.eclipse.smila.zookeeper.ZkConfigurationUpdateWatcher;
import org.eclipse.smila.zookeeper.ZooKeeperService;

/** Tests for {@link ZkConfigurationUpdateWatcher}. */
public class TestZkConfigurationUpdateWatcher extends DeclarativeServiceTestCase {

  private ZkConfigurationUpdateWatcher _watcher1;

  private MockService _service1;

  private ZkConfigurationUpdateWatcher _watcher2;

  private MockService _service2;

  @Override
  protected void setUp() throws Exception {
    super.setUp();
    final ZooKeeperService zkService = getService(ZooKeeperService.class);
    _watcher1 = new ZkConfigurationUpdateWatcher(zkService, "testing");
    _watcher2 = new ZkConfigurationUpdateWatcher(zkService, "testing");
    _service1 = new MockService(_watcher1);
    _service2 = new MockService(_watcher2);
  }

  @Override
  protected void tearDown() throws Exception {
    _watcher1.stopPolling();
    _watcher1.stopWatching();
    _watcher2.stopPolling();
    _watcher2.stopWatching();
    super.tearDown();
  }

  public void testWatchForConfigUpdate() throws Exception {
    final String configName = "testWatchForConfigUpdate";
    _watcher2.startWatching();
    _service1.update(configName);
    _service2.waitForNotification(configName, 1);
    _service1.update(configName);
    _service2.waitForNotification(configName, 2);
    _service1.update(configName);
    _service2.waitForNotification(configName, 3);
  }

  public void testWatchForConfigDelete() throws Exception {
    final String configName = "testWatchForConfigDelete";
    _watcher2.startWatching();
    _service1.update(configName + "1");
    _service1.update(configName + "2");
    _service2.waitForNotification(configName + "1", true);
    _service2.waitForNotification(configName + "2", true);
    _service1.delete(configName + "1");
    _service2.waitForNotification(configName + "1", false);
    assertTrue(_service2.getConfigNames().contains(configName + "2"));
  }

  public void testWatchForConfigLoadOnStart() throws Exception {
    final String configName = "testWatchForConfigLoadOnStart";
    _service2.loadOnStart(configName);
    _watcher2.startWatching();
    _service1.update(configName);
    _service2.waitForNotification(configName, 2);
  }

  public void testPollForConfigUpdate() throws Exception {
    final String configName = "testPollForConfigUpdate";
    _watcher2.startPolling(1);
    _service1.update(configName);
    _service2.waitForNotification(configName, true);
  }

  public void testPollForConfigDelete() throws Exception {
    final String configName = "testPollForConfigDelete";
    _watcher2.startPolling(1);
    _service1.update(configName + "1");
    _service1.update(configName + "2");
    _service2.waitForNotification(configName + "1", true);
    _service2.waitForNotification(configName + "2", true);
    _service1.delete(configName + "1");
    _service2.waitForNotification(configName + "1", false);
    assertTrue(_service2.getConfigNames().contains(configName + "2"));
  }

  public void testMultiDeleteAdd() throws Exception {
    final String configName = "testMultiDeleteAdd";
    final int noOfConfigs = 3;
    _watcher2.startWatching();
    for (int i = 0; i < noOfConfigs; i++) { // create configs 0, 1, 2
      _service1.update(configName + i);
    }
    for (int i = 0; i < noOfConfigs; i++) {
      _service2.waitForNotification(configName + i, 1);
    }
    assertFalse(_service2.getConfigNames().contains(configName + noOfConfigs));
    for (int i = 0; i < noOfConfigs; i++) { // delete configs 0, 1, 2
      _service1.delete(configName + i);
    }
    for (int i = 1; i <= noOfConfigs; i++) { // now create configs 1, 2, 3
      _service1.update(configName + i);
    }
    _service2.waitForNotification(configName + 0, false);
    for (int i = 1; i <= noOfConfigs; i++) {
      _service2.waitForNotification(configName + i, 1);
    }
  }

  private class MockService implements UpdateableService {

    private final Map<String, Integer> _configNames = new HashMap<>();

    private final ZkConfigurationUpdateWatcher _watcher;

    public MockService(final ZkConfigurationUpdateWatcher watcher) throws Exception {
      _watcher = watcher;
      _watcher.registerService(this);
      _watcher.initialize();
    }

    public Collection<String> getConfigNames() {
      return _configNames.keySet();
    }

    public void waitForNotification(final String configName, final boolean expectConfigExists) throws Exception {
      for (int i = 0; i < 50; i++) {
        if (expectConfigExists == _configNames.containsKey(configName)) {
          return;
        }
        Thread.sleep(100);
      }
      fail((expectConfigExists ? "Update" : "Delete") + " notification for '" + configName
        + "' not processed after 5 seconds.");
    }

    public void waitForNotification(final String configName, final Integer expectedUpdateCount) throws Exception {
      for (int i = 0; i < 50; i++) {
        if (expectedUpdateCount == _configNames.get(configName)) {
          return;
        }
        Thread.sleep(100);
      }
      fail("Not reached " + expectedUpdateCount + " updates for '" + configName
        + "' after 5 seconds, last count was " + _configNames.get(configName));
    }

    public void loadOnStart(final String configName) {
      updateInternal(configName);
      _watcher.configLoadedOnStart(configName, UUID.randomUUID().toString());
    }

    public void update(final String configName) {
      updateInternal(configName);
      _watcher.configUpdated(configName, UUID.randomUUID().toString());
    }

    private void updateInternal(final String configName) {
      if (_configNames.containsKey(configName)) {
        _configNames.put(configName, _configNames.get(configName) + 1);
      } else {
        _configNames.put(configName, 1);
      }
    }

    public void delete(final String configName) {
      _configNames.remove(configName);
      _watcher.configDeleted(configName);
    }

    @Override
    public void synchronizeConfiguration(final String configName, final boolean isDeleted) {
      if (isDeleted) {
        _configNames.remove(configName);
      } else {
        updateInternal(configName);
      }
    }
  }

}