/*
    * 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;
   }
 
 }