/* Stratosphere input split for Ptolemy tokens.
 * 
 * Copyright (c) 2014 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2015-08-24 22:42:20 +0000 (Mon, 24 Aug 2015) $' 
 * '$Revision: 33628 $'
 * 
 * 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.stratosphere.io.input;

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

import org.kepler.ddp.Utilities;

import eu.stratosphere.core.io.GenericInputSplit;
import ptolemy.data.StringToken;
import ptolemy.data.Token;

/** An input split for Ptolemy tokens. 
 * 
 * @author Daniel Crawl
 * @version $Id: TokenInputSplit.java 33628 2015-08-24 22:42:20Z crawl $
 * 
 */
public class TokenInputSplit extends GenericInputSplit {

        /** A constructor that takes no arguments is required. */
        public TokenInputSplit() {
        }
        
        /** Create a new TokenInputSplit.
         *  @param i the split number.
         *  @param token the token.
         */
        public TokenInputSplit(int i, Token token) {
                super(i);
                _token = token;
                //System.out.println("new split " + i + " for " + token);
        }

        /** Get the token. */
        public Token getToken() {
                return _token;
        }
        
        /** Write the token to the output stream. */
        @Override
        public void write(DataOutput out) throws IOException {

                super.write(out);
                
                // 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);
                }
                
                //System.out.println("done writing split for " + _token.stringValue());
                
        }

        /** Read a token from the input stream. */
        @Override
        public void read(DataInput in) throws IOException {
                
                super.read(in);
                
                // 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);

        }
        
        /** Get a string representation of this split. */
        @Override
        public String toString() {
            
            String tokenStr = null;
        if(_token instanceof StringToken) {
            tokenStr = ((StringToken)_token).stringValue();
        } else {
            tokenStr = _token.toString();
        }

                return "{ number: " + number + ", token: " + tokenStr + " }";
        }

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

}