/**
 * Copyright 2008 The Apache Software Foundation
 *
 * 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 net.sf.katta.zk;

import java.util.List;

import net.sf.katta.AbstractKattaTest;
import net.sf.katta.ZkServer;
import net.sf.katta.index.IndexMetaData;
import net.sf.katta.util.KattaException;
import net.sf.katta.util.ZkConfiguration;

import org.apache.hadoop.io.Text;

import com.yahoo.zookeeper.proto.WatcherEvent;

public class ZKClientTest extends AbstractKattaTest {

  static Integer _mutex = new Integer(-1);

  public void testWait() throws Exception {
    final ZKClient client = new ZKClient(conf);
    try {
      client.start(500);
      fail("this should fail, since no server is yet started.");
    } catch (final Exception e) {
      // expected
    }
    final ZkServer zkServer = new ZkServer(conf);
    client.start(3000);// now should work
    client.close();
    zkServer.shutdown();
  }

  public void testCreateFolder() throws KattaException, InterruptedException {
    final ZkServer zkServer = new ZkServer(conf);
    final ZKClient client = new ZKClient(conf);
    final String path = "/katta";
    client.start(10000);
    if (client.exists(path)) {
      assertTrue(client.deleteRecursive(path));
    }
    assertFalse(client.exists(path));
    client.create(path);
    assertTrue(client.exists(path));
    assertTrue(client.delete(path));
    final String value = "some value";
    client.create(path, new Text(value));
    final Text text = new Text();
    client.readData(path, text);
    assertEquals(value, text.toString());
    assertTrue(client.exists(path));
    assertTrue(client.delete(path));

    client.create(path);
    List<String> children = client.getChildren(path);
    assertEquals(0, children.size());

    client.create("/katta/child1");
    client.create("/katta/child2");
    children = client.getChildren(path);
    assertEquals(2, children.size());
    client.deleteRecursive(path);
    client.close();
    Thread.sleep(2000);
    zkServer.shutdown();
  }

  public void testChildNotifications() throws Exception {
    final ZkServer zkServer = new ZkServer(conf);
    final ZKClient client = new ZKClient(conf);
    client.start(10000);
    final MyListener listener = new MyListener();
    final String katta = "/katta";
    if (client.exists(katta)) {
      client.deleteRecursive(katta);
    }
    client.create(katta);
    client.subscribeChildChanges(katta, listener);
    for (int i = 0; i < 10; i++) {
      synchronized (_mutex) {
        client.create(katta + "/" + i);
        _mutex.wait();
      }
    }
    assertEquals(10, listener._counter);
    client.close();
    Thread.sleep(200);
    zkServer.shutdown();
  }

  public void testDataNotifications() throws Exception {
    final ZkServer zkServer = new ZkServer(conf);
    final ZKClient client = new ZKClient(conf);
    client.start(10000);
    final MyListener listener = new MyListener();
    final String katta = "/katta";
    if (client.exists(katta)) {
      client.deleteRecursive(katta);
    }
    client.create(katta, new IndexMetaData("path", "someAnalyzr", 3, IndexMetaData.IndexState.ANNOUNCED));
    client.subscribeDataChanges(katta, listener);
    for (int i = 0; i < 10; i++) {
      synchronized (_mutex) {
        final IndexMetaData indexMetaData = new IndexMetaData("path", "someAnalyzr" + i, 3,
            IndexMetaData.IndexState.ANNOUNCED);
        client.writeData(katta, indexMetaData);
        _mutex.wait();
      }
    }
    assertEquals(10, listener._counter);
    client.close();
    Thread.sleep(200);
    zkServer.shutdown();
  }

  public void testGetPath() throws Exception {
    final ZKClient client = new ZKClient(new ZkConfiguration());
    assertEquals("name", client.getNodeNameFromPath("/foo/bar/name"));
    assertEquals("name", client.getNodeNameFromPath("/foo/bar/name/"));
  }

  private class MyListener implements IZKEventListener {
    public int _counter = 0;

    public void process(final WatcherEvent event) {
      _counter++;
      synchronized (_mutex) {
        _mutex.notify();
      }

    }
  }
}

@@ -1278,16 +1278,132 @@
 r(-1);%0A%0A
+  @Override%0A  protected void onSetUp() throws Exception %7B%0A    System.out.println(%22_____________%22 + getName());%0A  %7D%0A%0A
   public
@@ -1560,16 +1560,19 @@
 ince no 
+zk 
 server i
@@ -1655,38 +1655,14 @@
     
-final ZkServer zkServer = new 
+create
 ZkSe
@@ -1662,28 +1662,24 @@
 ateZkServer(
-conf
 );%0A    clien
@@ -1735,33 +1735,8 @@
 ();%0A
-    zkServer.shutdown();%0A
   %7D%0A
@@ -1823,38 +1823,14 @@
     
-final ZkServer zkServer = new 
+create
 ZkSe
@@ -1826,36 +1826,32 @@
  createZkServer(
-conf
 );%0A    final ZKC
@@ -2758,57 +2758,8 @@
 ();%0A
-    Thread.sleep(2000);%0A    zkServer.shutdown();%0A
   %7D%0A
@@ -3003,39 +3003,42 @@
 inal String 
-katta = %22/katta
+file = %22/childFile
 %22;%0A    if (c
@@ -3042,37 +3042,36 @@
  (client.exists(
-katta
+file
 )) %7B%0A      clien
@@ -3080,37 +3080,36 @@
 deleteRecursive(
-katta
+file
 );%0A    %7D%0A    cli
@@ -3119,21 +3119,20 @@
 .create(
-katta
+file
 );%0A    c
@@ -3151,37 +3151,36 @@
 ibeChildChanges(
-katta
+file
 , listener);%0A   
@@ -3263,21 +3263,20 @@
 .create(
-katta
+file
  + %22/%22 +
@@ -3684,29 +3684,32 @@
 g katta = %22/
-k
+d
 at
-t
 a
+File
 %22;%0A    if (c

improve test stability of ZKClientTest