/*
 * Copyright 2009-2013 by The Regents of the University of California
 * Licensed 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 from
 * 
 *     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 edu.uci.ics.pregelix.example.data;

import edu.uci.ics.pregelix.api.graph.NormalizedKeyComputer;
import edu.uci.ics.pregelix.api.util.SerDeUtils;

/**
 * @author yingyib
 */
public class VLongNormalizedKeyComputer implements NormalizedKeyComputer {

    private static final int POSTIVE_LONG_MASK = (3 << 30);
    private static final int NON_NEGATIVE_INT_MASK = (2 << 30);
    private static final int NEGATIVE_LONG_MASK = (0 << 30);

    @Override
    public int getNormalizedKey(byte[] bytes, int start, int length) {
        try {
            long value = SerDeUtils.readVLong(bytes, start, length);
            int highValue = (int) (value >> 32);
            if (highValue > 0) {
                /**
                 * larger than Integer.MAX
                 */
                int highNmk = getKey(highValue);
                highNmk >>= 2;
                highNmk |= POSTIVE_LONG_MASK;
                return highNmk;
            } else if (highValue == 0) {
                /**
                 * smaller than Integer.MAX but >=0
                 */
                int lowNmk = (int) value;
                lowNmk >>= 2;
                lowNmk |= NON_NEGATIVE_INT_MASK;
                return lowNmk;
            } else {
                /**
                 * less than 0; TODO: have not optimized for that
                 */
                int highNmk = getKey(highValue);
                highNmk >>= 2;
                highNmk |= NEGATIVE_LONG_MASK;
                return highNmk;
            }
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
    }

    private int getKey(int value) {
        long unsignedFirstValue = (long) value;
        int nmk = (int) ((unsignedFirstValue - ((long) Integer.MIN_VALUE)) & 0xffffffffL);
        return nmk;
    }

}

@@ -248,18 +248,17 @@
  from%0A *
- 
 %0A
+
  *     h
@@ -301,17 +301,16 @@
 E-2.0%0A *
- 
 %0A * Unle
@@ -1308,21 +1308,33 @@
 if (
-highV
+v
 alue %3E 
-0
+Integer.MAX_VALUE
 ) %7B%0A
@@ -1601,22 +1601,48 @@
 if (
-highV
+v
 alue 
-=
+%3E
 = 0
+ && value %3C= Integer.MAX_VALUE
 ) %7B%0A
@@ -2355,15 +2355,8 @@
 ue =
- (long)
  val
@@ -2411,15 +2411,8 @@
  - (
-(long) 
 Inte

fix normalized key computer