content: /*
   * 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.engine.datasource.record;
  
  
  import java.io.IOException;
  import java.text.SimpleDateFormat;
  import java.util.Calendar;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.TreeMap;
  import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
  import org.apache.hadoop.chukwa.extraction.engine.Record;
  import org.apache.hadoop.chukwa.extraction.engine.SearchResult;
  import org.apache.hadoop.chukwa.extraction.engine.Token;
  import org.apache.hadoop.chukwa.extraction.engine.datasource.DataSource;
  import org.apache.hadoop.chukwa.extraction.engine.datasource.DataSourceException;
  import org.apache.hadoop.chukwa.inputtools.mdl.DataConfig;
  import org.apache.hadoop.fs.FileSystem;
  
  public class RecordDS implements DataSource {
  
    private static FileSystem fs = null;
    private static ChukwaConfiguration conf = null;
  
    private static String rootFolder = null;
    private static DataConfig dataConfig = null;
  
    static {
      dataConfig = new DataConfig();
      rootFolder = dataConfig.get("chukwa.engine.dsDirectory.rootFolder");
      conf = new ChukwaConfiguration();
      try {
        fs = FileSystem.get(conf);
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  
    public SearchResult search(SearchResult result, String cluster,
        String dataSource, long t0, long t1, String filter, Token token)
        throws DataSourceException {
  
      String filePath = rootFolder + "/" + cluster + "/" + dataSource;
  
      System.out.println("filePath [" + filePath + "]");
      Calendar calendar = Calendar.getInstance();
      calendar.setTimeInMillis(t1);
  
      TreeMap<Long, List<Record>> records = result.getRecords();
      int maxCount = 200;
      SimpleDateFormat sdf = new java.text.SimpleDateFormat("_yyyyMMdd_HH_");
      do {
        System.out.println("start Date [" + calendar.getTime() + "]");
        String fileName = sdf.format(calendar.getTime());
        int minutes = calendar.get(Calendar.MINUTE);
        int dec = minutes / 10;
        fileName += dec;
  
        int m = minutes - (dec * 10);
        if (m < 5) {
          fileName += "0.1.evt";
        } else {
          fileName += "5.1.evt";
        }
  
        fileName = filePath + "/" + dataSource + fileName;
  
        // System.out.println("JB fileName  [" +fileName + "]");
  
        try {
          System.out.println("BEFORE fileName  [" + fileName + "]");
  
          // List<Record> evts =
          // ChukwaFileParser.readData(cluster,dataSource,maxCount, t1, t0,
          // Long.MAX_VALUE, filter, fileName, fs);
          List<Record> evts = ChukwaSequenceFileParser.readData(cluster,
              dataSource, maxCount, t1, t0, Long.MAX_VALUE, filter, fileName, fs,
              conf);
  
          maxCount = maxCount - evts.size();
          System.out.println("AFTER fileName  [" + fileName + "] count="
              + evts.size() + " maxCount=" + maxCount);
          for (Record evt : evts) {
            System.out.println("AFTER Loop  [" + evt.toString() + "]");
            long timestamp = evt.getTime();
            if (records.containsKey(timestamp)) {
              records.get(timestamp).add(evt);
            } else {
              List<Record> list = new LinkedList<Record>();
              list.add(evt);
              records.put(timestamp, list);
            }
          }
        } catch (Exception e) {
          e.printStackTrace();
        }
  
        if (maxCount <= 0) {
          System.out.println("BREAKING LOOP AFTER [" + fileName + "] maxCount="
              + maxCount);
          break;
        }
  
        calendar.add(Calendar.MINUTE, -5);
  
        System.out.println("calendar  [" + calendar.getTimeInMillis() + "] ");
        System.out.println("end       [" + (t0 - 1000 * 60 * 5) + "] ");
      } while (calendar.getTimeInMillis() > (t0 - 1000 * 60 * 5)); // <= need some
                                                                   // code here
      // Need more than this to compute the end
  
      return result;
    }
  
    public boolean isThreadSafe() {
      return true;
    }
  }
  