/* Hadoop InputSplit for Ptolemy tokens.
 * 
 * Copyright (c) 2014 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2014-07-02 15:58:19 +0000 (Wed, 02 Jul 2014) $' 
 * '$Revision: 32804 $'
 * 
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */
package org.kepler.hadoop.io.input;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.kepler.ddp.Utilities;
import org.kepler.hadoop.io.TokenWritable;

import ptolemy.data.StringToken;
import ptolemy.data.Token;

/** Hadoop InputSplit for Ptolemy tokens.
 * 
 *  @author Daniel Crawl
 *  @verion $Id: TokenInputSplit.java 32804 2014-07-02 15:58:19Z crawl $
 */
public class TokenInputSplit extends InputSplit implements Writable {

    /** No argument constructor is required. */
    public TokenInputSplit() {
    }
    
    /** Create a TokenInputSplit for a token. */
    public TokenInputSplit(Token token) {
        _token = token;
    }

    /** Get the number of bytes in the split. */
    @Override
    public long getLength() throws IOException, InterruptedException {
        
        String tokenStr = null;
        if(_token instanceof StringToken) {
            tokenStr = ((StringToken)_token).stringValue();
        } else {
            tokenStr = _token.toString();
        }
        return new Text(tokenStr).getLength();
    }

    /** Get the host locations of the split. In this class, returns none. */
    @Override
    public String[] getLocations() throws IOException, InterruptedException {
        return new String[] {};
    }

    /** Get a Writable for this split. */
    public TokenWritable getTokenWritable() {
        return new TokenWritable(_token);
    }


    /** Unserialize the fields for this split. */
    @Override
    public void readFields(DataInput in) throws IOException {
        
        // TODO copied from Stratosphere TokenInputSplit

        // read token type
        int typeVal = in.readUnsignedByte();
        
        if (typeVal >= HIGH_BIT) {
            int shift = 7;
            int curr;
            typeVal = typeVal & 0x7f;
            while ((curr = in.readUnsignedByte()) >= HIGH_BIT) {
                typeVal |= (curr & 0x7f) << shift;
                shift += 7;
            }
            typeVal |= curr << shift;
        }
        
        Utilities.TokenType type = Utilities.TokenType.getInstance(typeVal);

        // XXX copied from StringValue
        
        int len = in.readUnsignedByte();

        if (len >= HIGH_BIT) {
            int shift = 7;
            int curr;
            len = len & 0x7f;
            while ((curr = in.readUnsignedByte()) >= HIGH_BIT) {
                len |= (curr & 0x7f) << shift;
                shift += 7;
            }
            len |= curr << shift;
        }
        
        //this.len = len;
        //this.hashCode = 0;
        //ensureSize(len);
        //final char[] data = this.value;
        final char[] data = new char[len];

        for (int i = 0; i < len; i++) {
            int c = in.readUnsignedByte();
            if (c < HIGH_BIT)
                data[i] = (char) c;
            else {
                int shift = 7;
                int curr;
                c = c & 0x7f;
                while ((curr = in.readUnsignedByte()) >= HIGH_BIT) {
                    c |= (curr & 0x7f) << shift;
                    shift += 7;
                }
                c |= curr << shift;
                data[i] = (char) c;
            }
        }
        
        String tokenStr = String.valueOf(data);
        _token = Utilities.createTokenFromString(tokenStr, type);
        
    }

    /** Serialize the fields in this split. */
    @Override
    public void write(DataOutput out) throws IOException {
        
        // TODO copied from Stratosphere TokenInputSplit
        
        // write the token type
        final Utilities.TokenType type = Utilities.getTokenTypeForSerialization(_token);
        int typeVal = type.getValue();
        
        while(typeVal >= HIGH_BIT) {
            out.write(typeVal | HIGH_BIT);
            typeVal >>>= 7;
        }
        out.write(typeVal);
        
        // XXX copied from StringValue
        
        String tokenStr = null;
        if(_token instanceof StringToken) {
            tokenStr = ((StringToken)_token).stringValue();
        } else {
            tokenStr = _token.toString();
        }
        final int length = tokenStr.length();

        int len = length;
        // write the length, variable-length encoded
        while (len >= HIGH_BIT) {
            out.write(len | HIGH_BIT);
            len >>>= 7;
        }
        out.write(len);

        // write the char data, variable length encoded
        for (int i = 0; i < len; i++) {
            int c = tokenStr.charAt(i);

            while (c >= HIGH_BIT) {
                out.write(c | HIGH_BIT);
                c >>>= 7;
            }
            out.write(c);
        }
    }

    /** The token for this split. */
    private Token _token;
    
    private static final int HIGH_BIT = 0x1 << 7;

}