core/src/main/java/org/apache/hadoop/chukwa/extraction/demux/processor/mapper/AbstractProcessor.java

/*
 * 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.extraction.demux.processor.mapper;


import java.nio.charset.Charset;
import java.util.Calendar;

import org.apache.hadoop.chukwa.ChukwaArchiveKey;
import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
import org.apache.hadoop.chukwa.extraction.engine.Record;
import org.apache.hadoop.chukwa.util.RecordConstants;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.log4j.Logger;

public abstract class AbstractProcessor implements MapProcessor {
  static Logger log = Logger.getLogger(AbstractProcessor.class);

  Calendar calendar = Calendar.getInstance();
  byte[] bytes;
  int[] recordOffsets;
  protected int currentPos = 0;
  protected int startOffset = 0;

  protected ChukwaArchiveKey archiveKey = null;
  protected ChukwaRecordKey key = new ChukwaRecordKey();
  protected Chunk chunk = null;

  boolean chunkInErrorSaved = false;
  OutputCollector<ChukwaRecordKey, ChukwaRecord> output = null;
  Reporter reporter = null;

  public AbstractProcessor() {
  }

  protected abstract void parse(String recordEntry,
      OutputCollector<ChukwaRecordKey, ChukwaRecord> output, Reporter reporter)
      throws Throwable;

  protected void saveChunkInError(Throwable throwable) {
    if (chunkInErrorSaved == false) {
      try {
        ChunkSaver.saveChunk(chunk, throwable, output, reporter);
        chunkInErrorSaved = true;
      } catch (Exception e) {
        e.printStackTrace();
      }
    }

  }

  public void process(ChukwaArchiveKey archiveKey, Chunk chunk,
      OutputCollector<ChukwaRecordKey, ChukwaRecord> output, Reporter reporter) {
    chunkInErrorSaved = false;

    this.archiveKey = archiveKey;
    this.output = output;
    this.reporter = reporter;

    reset(chunk);

    while (hasNext()) {
      try {
        parse(nextLine(), output, reporter);
      } catch (Throwable e) {
        saveChunkInError(e);
      }
    }
  }

  protected void buildGenericRecord(ChukwaRecord record, String body,
      long timestamp, String dataSource) {
    calendar.setTimeInMillis(timestamp);
    calendar.set(Calendar.MINUTE, 0);
    calendar.set(Calendar.SECOND, 0);
    calendar.set(Calendar.MILLISECOND, 0);

    key.setKey("" + calendar.getTimeInMillis() + "/" + chunk.getSource() + "/"
        + timestamp);
    key.setReduceType(dataSource);

    if (body != null) {
      record.add(Record.bodyField, body);
    }
    record.setTime(timestamp);

    record.add(Record.tagsField, chunk.getTags());
    record.add(Record.sourceField, chunk.getSource());
    record.add(Record.applicationField, chunk.getStreamName());

  }

  protected void reset(Chunk chunk) {
    this.chunk = chunk;
    this.bytes = chunk.getData();
    this.recordOffsets = chunk.getRecordOffsets();
    currentPos = 0;
    startOffset = 0;
  }

  protected boolean hasNext() {
    return (currentPos < recordOffsets.length);
  }

  protected String nextLine() {
    String log = new String(bytes, startOffset, (recordOffsets[currentPos]
        - startOffset + 1), Charset.forName("UTF-8"));
    startOffset = recordOffsets[currentPos] + 1;
    currentPos++;
    return RecordConstants.recoverRecordSeparators("\n", log);
  }

public int getCurrentPos() {
	return currentPos;
}

public void setCurrentPos(int currentPos) {
	this.currentPos = currentPos;
}

public int getStartOffset() {
	return startOffset;
}

public void setStartOffset(int startOffset) {
	this.startOffset = startOffset;
}  
}