package org.apache.helix.manager.zk;
   
   /*
    * 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.
   */
  
  import java.io.ByteArrayInputStream;
  import java.io.StringWriter;
  import java.util.List;
  import java.util.Map;
  
  import org.I0Itec.zkclient.serialize.ZkSerializer;
  import org.apache.helix.HelixException;
  import org.apache.helix.ZNRecord;
  import org.apache.log4j.Logger;
  import org.codehaus.jackson.map.DeserializationConfig;
  import org.codehaus.jackson.map.ObjectMapper;
  import org.codehaus.jackson.map.SerializationConfig;
  
  
  public class ZNRecordSerializer implements ZkSerializer
  {
    private static Logger logger = Logger.getLogger(ZNRecordSerializer.class);
  
    private static int getListFieldBound(ZNRecord record)
    {
      int max = Integer.MAX_VALUE;
      if (record.getSimpleFields().containsKey(ZNRecord.LIST_FIELD_BOUND))
      {
        String maxStr = record.getSimpleField(ZNRecord.LIST_FIELD_BOUND);
        try
        {
          max = Integer.parseInt(maxStr);
        }
        catch (Exception e)
        {
          logger.error("IllegalNumberFormat for list field bound: " + maxStr);
        }
      }
      return max;
    }
  
    @Override
    public byte[] serialize(Object data)
    {
      if (!(data instanceof ZNRecord))
      {
        // null is NOT an instance of any class
        logger.error("Input object must be of type ZNRecord but it is " + data + ". Will not write to zk");
        throw new HelixException("Input object is not of type ZNRecord (was " + data + ")");
      }
  
      ZNRecord record = (ZNRecord) data;
      
      // apply retention policy
      int max = getListFieldBound(record);
      if (max < Integer.MAX_VALUE)
      {
        Map<String, List<String>> listMap = record.getListFields();
        for (String key : listMap.keySet())
        {
          List<String> list = listMap.get(key);
          if (list.size() > max)
          {
            listMap.put(key, list.subList(0, max));
          }
        }
      }
  
      // do serialization
      ObjectMapper mapper = new ObjectMapper();
      SerializationConfig serializationConfig = mapper.getSerializationConfig();
      serializationConfig.set(SerializationConfig.Feature.INDENT_OUTPUT, true);
      serializationConfig.set(SerializationConfig.Feature.AUTO_DETECT_FIELDS, true);
      serializationConfig.set(SerializationConfig.Feature.CAN_OVERRIDE_ACCESS_MODIFIERS, true);
      StringWriter sw = new StringWriter();
      try
      {
        mapper.writeValue(sw, data);
      } catch (Exception e)
      {
        logger.error("Exception during data serialization. Will not write to zk. Data (first 1k): "
            + sw.toString().substring(0, 1024), e);
        throw new HelixException(e);
     }
     
     if (sw.toString().getBytes().length > ZNRecord.SIZE_LIMIT)
     {
       logger.error("Data size larger than 1M, ZNRecord.id: " + record.getId() 
           + ". Will not write to zk. Data (first 1k): " + sw.toString().substring(0, 1024));
       throw new HelixException("Data size larger than 1M, ZNRecord.id: " + record.getId());
     }
     return sw.toString().getBytes();
   }
 
   @Override
   public Object deserialize(byte[] bytes)
   {
     if (bytes == null || bytes.length == 0)
     {
       // reading a parent/null node
       return null;
     }
 
     ObjectMapper mapper = new ObjectMapper();
     ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
 
     DeserializationConfig deserializationConfig = mapper.getDeserializationConfig();
     deserializationConfig.set(DeserializationConfig.Feature.AUTO_DETECT_FIELDS, true);
     deserializationConfig.set(DeserializationConfig.Feature.AUTO_DETECT_SETTERS, true);
     deserializationConfig.set(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, true);
     try
     {
       ZNRecord zn = mapper.readValue(bais, ZNRecord.class);
       return zn;
     } catch (Exception e)
     {
       logger.error("Exception during deserialization of bytes: " + new String(bytes), e);
       return null;
     }
   }
 }