CustomLockInternalsDriver.java

package quux00.curator;

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.LockInternalsDriver;
import org.apache.curator.framework.recipes.locks.PredicateResults;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

public class CustomLockInternalsDriver implements LockInternalsDriver {

  private String mylockPath;
  
  public String getOwnLockPath() {
    return mylockPath;
  }
  
  @Override
  public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception {
    System.out.printf("DEBUG +1: %s | %s\n", path, lockNodeBytes);

    String ourPath;
    if (lockNodeBytes != null) {
      System.out.println("DEBUG +1b");
      ourPath = client.create().creatingParentsIfNeeded().withProtection().
          withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
          forPath(path, lockNodeBytes);
    } else {
      System.out.println("DEBUG +1c");
      ourPath = client.create().creatingParentsIfNeeded().withProtection().
          withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
          forPath(path);
    }
    System.out.println("DEBUG +2: " + ourPath);
    mylockPath = ourPath;
    return ourPath;
  }

  @Override
  public PredicateResults getsTheLock(CuratorFramework client, List<String> children, 
      String sequenceNodeName, int maxLeases) throws Exception {
    
    System.out.printf("DEBUG +3: %s | %s | %d\n", children, sequenceNodeName, maxLeases);
    
    int ourIndex = children.indexOf(sequenceNodeName);
    validateOurIndex(sequenceNodeName, ourIndex);

    boolean getsTheLock = ourIndex < maxLeases;
    String pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);
    System.out.println("DEBUG +4: " + pathToWatch);

    // DEBUG
    pathToWatch = mylockPath;
    System.out.println("LockInternalsDirver: setting pathToWatch to " + "/XXX" + pathToWatch);
    // END DEBUG
    
    return new PredicateResults("/XXX" + pathToWatch, getsTheLock);
  }
  
  @Override
  public String fixForSorting(String str, String lockName) {
    return standardFixForSorting(str, lockName);
  }

  public static String standardFixForSorting(String str, String lockName) {
    int index = str.lastIndexOf(lockName);
    if ( index >= 0 ) {
      index += lockName.length();
      return index <= str.length() ? str.substring(index) : "";
    }
    return str;
  }

  static void validateOurIndex(String sequenceNodeName, int ourIndex) throws KeeperException {
    if ( ourIndex < 0 ) {
      throw new KeeperException.NoNodeException("Sequential path not found: " + sequenceNodeName);
    }
  }
}