/**
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 io.hops.metadata.hdfs.entity;

import com.google.common.base.Charsets;
import io.hops.metadata.common.FinderType;

import java.util.Objects;

public final class StoredXAttr {
  
  public static final int MAX_NUM_USER_XATTRS_PER_INODE = 127;
  public static final int MAX_NUM_SYS_XATTRS_PER_INODE = 127;
  public static final int MAX_XATTR_NAME_SIZE = 255;
  public static final int MAX_XATTR_VALUE_SIZE = 13500;
  
  public enum Finder implements FinderType<StoredXAttr> {
    ByPrimaryKey,
    ByPrimaryKeyBatch,
    ByInodeId;
    
    @Override
    public Class getType() {
      return StoredXAttr.class;
    }
    
    @Override
    public Annotation getAnnotated() {
      switch (this){
        case ByPrimaryKey:
          return Annotation.PrimaryKey;
        case ByPrimaryKeyBatch:
          return Annotation.Batched;
        case ByInodeId:
          return Annotation.PrunedIndexScan;
        default:
          throw new IllegalStateException();
      }
    }
  }
  
  public final static class PrimaryKey{
    private final long inodeId;
    private final byte namespace;
    private final String name;
  
    public PrimaryKey(long inodeId, byte namespace, String name) {
      this.inodeId = inodeId;
      this.namespace = namespace;
      this.name = name;
    }
  
    public long getInodeId() {
      return inodeId;
    }
  
    public byte getNamespace() {
      return namespace;
    }
  
    public String getName() {
      return name;
    }
  
    @Override
    public boolean equals(Object o) {
      if (this == o) {
        return true;
      }
      if (!(o instanceof PrimaryKey)) {
        return false;
      }
      PrimaryKey that = (PrimaryKey) o;
      return inodeId == that.inodeId &&
          namespace == that.namespace &&
          name.equals(that.name);
    }
  
    @Override
    public int hashCode() {
      return Objects.hash(inodeId, namespace, name);
    }
  
    @Override
    public String toString() {
      return "PrimaryKey{" +
          "inodeId=" + inodeId +
          ", namespace=" + namespace +
          ", name='" + name + '\'' +
          '}';
    }
  }
  
  private final PrimaryKey primaryKey;
  private final String value;
  
  public StoredXAttr(long inodeId, byte namespace, String name, String value) {
    this.primaryKey = new PrimaryKey(inodeId, namespace, name);
    this.value = value;
  }
  
  public StoredXAttr(long inodeId, byte namespace, String name, byte[] value) {
    this(inodeId, namespace, name, getXAttrString(value));
  }
  
  public long getInodeId() {
    return primaryKey.getInodeId();
  }
  
  public byte getNamespace() {
    return primaryKey.getNamespace();
  }
  
  public String getName() {
    return primaryKey.getName();
  }
  
  public String getValue() {
    return value;
  }
  
  public PrimaryKey getPrimaryKey(){
    return primaryKey;
  }
  
  public byte[] getValueBytes(){
    return getXAttrBytes(value);
  }
  
  public final static byte[] getXAttrBytes(String val){
    if(val == null)
      return null;
    return val.getBytes(Charsets.UTF_8);
  }
  
  public final static String getXAttrString(byte[] val){
    if(val == null)
      return null;
    return new String(val,Charsets.UTF_8);
  }
  
  @Override
  public boolean equals(Object o) {
    if (this == o) {
      return true;
    }
    if (!(o instanceof StoredXAttr)) {
      return false;
    }
    StoredXAttr that = (StoredXAttr) o;
    return primaryKey.equals(that.primaryKey) &&
        getValue().equals(that.getValue());
  }
  
  @Override
  public int hashCode() {
    return Objects.hash(primaryKey, getValue());
  }
  
  @Override
  public String toString() {
    return "StoredXAttr{" +
        "primaryKey=" + primaryKey +
        ", value='" + value + '\'' +
        '}';
  }
}

@@ -927,16 +927,41 @@
 rType;%0A%0A
+import java.util.Arrays;%0A
 import j
@@ -1249,16 +1249,91 @@
  13500;%0A
+  public static final byte%5B%5D NON_EXISTENT_XATRR_VALUE = getXAttrBytes(%22%22);%0A
   %0A  pub
@@ -3092,22 +3092,22 @@
 e final 
-String
+byte%5B%5D
  value;%0A
@@ -3173,22 +3173,22 @@
 g name, 
-String
+byte%5B%5D
  value) 
@@ -3288,154 +3288,8 @@
 %0A  %0A
-  public StoredXAttr(long inodeId, byte namespace, String name, byte%5B%5D value) %7B%0A    this(inodeId, namespace, name, getXAttrString(value));%0A  %7D%0A  %0A
   pu
@@ -3505,30 +3505,30 @@
   %0A  public 
-String
+byte%5B%5D
  getValue() 
@@ -3625,81 +3625,8 @@
 %0A  %0A
-  public byte%5B%5D getValueBytes()%7B%0A    return getXAttrBytes(value);%0A  %7D%0A  %0A
   pu
@@ -3902,32 +3902,188 @@
 .UTF_8);%0A  %7D%0A  %0A
+  public static boolean xAttrExists(byte%5B%5D value)%7B%0A    if(value != null)%0A      return !Arrays.equals(NON_EXISTENT_XATRR_VALUE, value);%0A    return true;%0A  %7D%0A
   @Override%0A  pu

Use varbinary for xattrs value