/*
 * 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.
 */

package org.apache.hadoop.chukwa;


import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.Charset;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.chukwa.datacollection.DataFactory;
import org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor;

public class ChunkImpl implements org.apache.hadoop.io.Writable, Chunk {
  public final static int PROTOCOL_VERSION = 1;

  protected DataFactory dataFactory = DataFactory.getInstance();
  private String source = "";
  private String streamName = "";
  private String dataType = "";
  private String tags = "";
  private byte[] data = null;
  private int[] recordEndOffsets;
  private int protocolVersion = 1;
  private String debuggingInfo = "";

  private transient Adaptor initiator;
  long seqID;

  private static String localHostAddr;
  static {
    try {
      setHostAddress(InetAddress.getLocalHost().getHostName().toLowerCase());
    } catch (UnknownHostException e) {
      setHostAddress("localhost");
    }
  }
  
  public static void setHostAddress(String host) {
    ChunkImpl.localHostAddr = host;
  }
  
  
  public static ChunkImpl getBlankChunk() {
    return new ChunkImpl();
  }

  ChunkImpl() {
  }

  public ChunkImpl(String dataType, String streamName, long seq, byte[] data,
                   Adaptor source) {
    this.seqID = seq;
    this.source = localHostAddr;
    this.tags = dataFactory.getDefaultTags();
    this.streamName = streamName;
    this.dataType = dataType;
    this.data = (byte[]) data.clone();
    this.initiator = source;
  }

  /**
   * @see org.apache.hadoop.chukwa.Chunk#getData()
   */
  public byte[] getData() {
    return data.clone();
  }

  /**
   * @see org.apache.hadoop.chukwa.Chunk#setData(byte[])
   */
  public void setData(byte[] logEvent) {
    this.data = (byte[]) logEvent.clone();
  }

  /**
   * @see org.apache.hadoop.chukwa.Chunk#getStreamName()
   */
  public String getStreamName() {
    return streamName;
  }

  public void setStreamName(String logApplication) {
    this.streamName = logApplication;
  }

  public String getSource() {
    return source;
  }

  public void setSource(String logSource) {
    this.source = logSource;
  }

  public String getDebugInfo() {
    return debuggingInfo;
  }

  public void setDebugInfo(String a) {
    this.debuggingInfo = a;
  }

  /**
   * @see org.apache.hadoop.chukwa.Chunk#getSeqID()
   */
  public long getSeqID() {
    return seqID;
  }

  public void setSeqID(long l) {
    seqID = l;
  }

  public int getProtocolVersion() {
    return protocolVersion;
  }

  public void setProtocolVersion(int pv) {
    this.protocolVersion = pv;
  }

  public Adaptor getInitiator() {
    return initiator;
  }

  public void setInitiator(Adaptor a) {
    initiator = a;
  }

  public void setLogSource() {
    source = localHostAddr;
  }

  public int[] getRecordOffsets() {
    if (recordEndOffsets == null)
      recordEndOffsets = new int[] { data.length - 1 };
    return recordEndOffsets.clone();
  }

  public void setRecordOffsets(int[] offsets) {
    recordEndOffsets = (int[]) offsets.clone();
  }

  public String getDataType() {
    return dataType;
  }

  public void setDataType(String t) {
    dataType = t;
  }

  @Override
  public void addTag(String tags) {
    this.tags += " "+ tags;
  }

  /**
   * @see org.apache.hadoop.chukwa.Chunk#getTags()
   */
  public String getTags() {
    return tags;
  }
  
  /**
   * @see org.apache.hadoop.chukwa.Chunk#getTag(java.lang.String)
   */
  public String getTag(String tagName) {
    Pattern tagPattern = Pattern.compile("\\b"+tagName+"=\"([^\"]*)\"");
    if (tags != null) {
      Matcher matcher = tagPattern.matcher(tags);
      if (matcher.find()) {
        return matcher.group(1);
      }
    }
    return null;
  }
  
  /**
   * @see org.apache.hadoop.io.Writable#readFields(java.io.DataInput)
   */
  public void readFields(DataInput in) throws IOException {
    setProtocolVersion(in.readInt());
    if (protocolVersion != PROTOCOL_VERSION) {
      throw new IOException(
          "Protocol version mismatched, drop data.  source version: "
              + protocolVersion + ", collector version:" + PROTOCOL_VERSION);
    }
    setSeqID(in.readLong());
    setSource(in.readUTF());
    tags = in.readUTF(); // no public set method here
    setStreamName(in.readUTF());
    setDataType(in.readUTF());
    setDebugInfo(in.readUTF());

    int numRecords = in.readInt();
    recordEndOffsets = new int[numRecords];
    for (int i = 0; i < numRecords; ++i)
      recordEndOffsets[i] = in.readInt();
    data = new byte[recordEndOffsets[recordEndOffsets.length - 1] + 1];
    in.readFully(data);

  }

  /**
   * @see org.apache.hadoop.io.Writable#write(java.io.DataOutput)
   */
  public void write(DataOutput out) throws IOException {
    out.writeInt(PROTOCOL_VERSION);
    out.writeLong(seqID);
    out.writeUTF(source);
    out.writeUTF(tags);
    out.writeUTF(streamName);
    out.writeUTF(dataType);
    out.writeUTF(debuggingInfo);

    if (recordEndOffsets == null)
      recordEndOffsets = new int[] { data.length - 1 };

    out.writeInt(recordEndOffsets.length);
    for (int i = 0; i < recordEndOffsets.length; ++i)
      out.writeInt(recordEndOffsets[i]);

    out.write(data, 0, recordEndOffsets[recordEndOffsets.length - 1] + 1); 
    // byte at last offset is valid
  }

  public static ChunkImpl read(DataInput in) throws IOException {
    ChunkImpl w = new ChunkImpl();
    w.readFields(in);
    return w;
  }

  public String toString() {
    StringBuilder buffer = new StringBuilder();
    buffer.append(source);
    buffer.append(":");
    buffer.append(streamName);
    buffer.append(new String(data, Charset.forName("UTF-8")));
    buffer.append("/");
    buffer.append(seqID);
    return buffer.toString();
  }


  /**
   * @see org.apache.hadoop.chukwa.Chunk#getSerializedSizeEstimate()
   */
  public int getSerializedSizeEstimate() {
    int size = 2 * (source.length() + streamName.length() + dataType.length() 
        + debuggingInfo.length()); // length of strings (pessimistic)
    size += data.length + 4;
    if (recordEndOffsets == null)
      size += 8;
    else
      size += 4 * (recordEndOffsets.length + 1); // +1 for length of array
    size += 8; // uuid
    return size;
  }

  public void setRecordOffsets(java.util.Collection<Integer> carriageReturns) {
    recordEndOffsets = new int[carriageReturns.size()];
    int i = 0;
    for (Integer offset : carriageReturns)
      recordEndOffsets[i++] = offset;
  }
  
  public int getLength() {
    return data.length;
  }

}