package com.linkedin.helix.manager.zk;

import java.util.ArrayList;
import java.util.List;

import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;

public class ChainedPathZkSerializer implements PathBasedZkSerializer
{

  public static class Builder
  {
    private final ZkSerializer _defaultSerializer;
    private List<ChainItem> _items = new ArrayList<ChainItem>();

    private Builder(ZkSerializer defaultSerializer)
    {
      _defaultSerializer = defaultSerializer;
    }

    /**
     * Add a serializing strategy for the given path prefix
     */
    public Builder serialize(String path, ZkSerializer withSerializer)
    {
      _items.add(new ChainItem(path, withSerializer));
      return this;
    }
    
    /**
     * Builds the serializer with the given strategies and default serializer.
     */
    public ChainedPathZkSerializer build() {
      return new ChainedPathZkSerializer(_defaultSerializer, _items);
    }
  }
  
  /**
   * Create a builder that will use the given serializer by default
   * if no other strategy is given to solve the path in question.
   */
  public static Builder builder(ZkSerializer defaultSerializer) 
  {
    return new Builder(defaultSerializer);
  }

  private final List<ChainItem> _items;
  private final ZkSerializer _defaultSerializer;

  public ChainedPathZkSerializer(ZkSerializer defaultSerializer, List<ChainItem> items)
  {
    _items = items;
    _defaultSerializer = defaultSerializer;
  }

  @Override
  public byte[] serialize(Object data, String path) throws ZkMarshallingError
  {
    for (ChainItem item : _items)
    {
      if (item.matches(path)) return item._serializer.serialize(data);
    }
    return _defaultSerializer.serialize(data);
  }

  @Override
  public Object deserialize(byte[] bytes, String path)
      throws ZkMarshallingError
  {
    for (ChainItem item : _items)
    {
      if (item.matches(path)) return item._serializer.deserialize(bytes);
    }
    return _defaultSerializer.deserialize(bytes);
  }

  private static class ChainItem
  {
    final String _path;
    final ZkSerializer _serializer;

    ChainItem(String path, ZkSerializer serializer)
    {
      _path = path;
      _serializer = serializer;
    }

    boolean matches(String path)
    {
      if (_path.equals(path))
      {
        return true;
      } 
      else if (path.length() > _path.length())
      {
        if (path.startsWith(_path) && path.charAt(_path.length()) == '/') 
        {
          return true;
        }
      }
      return false;
    }
  }

}

@@ -61,16 +61,46 @@
 ayList;%0A
+import java.util.Collections;%0A
 import j
@@ -627,16 +627,141 @@
  prefix%0A
+     * The most specific path will triumph over a more generic (shorter)%0A     * one regardless of the ordering of the calls.%0A
      */%0A
@@ -868,20 +868,31 @@
 ainItem(
+normalize(
 path
+)
 , withSe
@@ -1512,21 +1512,22 @@
 er;%0A%0A  p
-ublic
+rivate
  Chained
@@ -1622,16 +1622,191 @@
  items;%0A
+    // sort by longest paths first%0A    // if two items would match one would be prefix of the other%0A    // and the longest must be more specific%0A    Collections.sort(_items);%0A
     _def
@@ -2419,16 +2419,49 @@
 hainItem
+ implements Comparable%3CChainItem%3E
 %0A  %7B%0A   
@@ -2944,20 +2944,416 @@
 false;%0A    %7D
+%0A%0A    @Override%0A    public int compareTo(ChainItem o)%0A    %7B%0A      return o._path.length() - _path.length();%0A    %7D%0A  %7D%0A  %0A  private static String normalize(String path) %7B%0A    if (!path.startsWith(%22/%22)) %7B%0A      // ensure leading slash%0A      path = %22/%22 + path;%0A    %7D%0A    if (path.endsWith(%22/%22)) %7B%0A      // remove trailing slash%0A      path = path.substring(0, path.length()-1);%0A    %7D%0A    return path;
 %0A  %7D%0A%0A%7D%0A

normalize path + specificity precedence