core/src/test/java/org/apache/hadoop/chukwa/extraction/demux/TestDemuxManager.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;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import junit.framework.Assert;
import junit.framework.TestCase;

import org.apache.hadoop.chukwa.conf.ChukwaConfiguration;
import org.apache.hadoop.chukwa.extraction.CHUKWA_CONSTANT;

public class TestDemuxManager extends TestCase {

  /**
   * Standard workflow
   */
  public void testScenario1() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    cc.set(CHUKWA_CONSTANT.HDFS_DEFAULT_NAME_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );
    
     
    try {
      
      File dataSinkDirectory = new File(chukwaRootDir +"/logs");
      dataSinkDirectory.mkdirs();
      File dataSinkFile = new File(chukwaRootDir +"/logs"+ "/dataSink1.done");
      dataSinkFile.createNewFile();
      
      DemuxManagerScenario dm = new DemuxManagerScenario1(cc,0);  
      dm.start();
      
      List<String> requireActions = new ArrayList<String>();
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:false");
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true");
      requireActions.add("runDemux:true");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true");
      requireActions.add("processData done");
      
      List<String> actions = dm.actions;
     
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for(int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

    } catch (Exception e) {
      e.printStackTrace();
     Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
  }
  
  /**
   * No dataSink file at  startup
   * Add one later
   */
  public void testScenario2() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    cc.set(CHUKWA_CONSTANT.HDFS_DEFAULT_NAME_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );

    try {
      DemuxManagerScenario dm = new DemuxManagerScenario2(cc);  
      dm.start();
 
      List<String> requireActions = new ArrayList<String>();
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:false");
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:false");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:false");
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true");
      requireActions.add("runDemux:true");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true");
      requireActions.add("processData done");

      List<String> actions = dm.actions;
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for(int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

    } catch (Exception e) {
      e.printStackTrace();
      Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
  }
  
  
  /**
   * DataSink file in dataSink directory
   * MR_INPUT_DIR already there
   * MR_INPUT_DIR should be reprocessed first
   * Then dataSink file
   */
  public void testScenario3() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    cc.set(CHUKWA_CONSTANT.HDFS_DEFAULT_NAME_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );

    try {
      File mrInputDir = new File(chukwaRootDir + CHUKWA_CONSTANT.DEFAULT_DEMUX_PROCESSING_DIR_NAME+ CHUKWA_CONSTANT.DEFAULT_DEMUX_MR_INPUT_DIR_NAME);
      mrInputDir.mkdirs();
      File dataSinkDirectory = new File(chukwaRootDir +"/logs");
      dataSinkDirectory.mkdirs();
      File dataSinkFile = new File(chukwaRootDir +"/logs"+ "/dataSink3.done");
      dataSinkFile.createNewFile();
      
      DemuxManagerScenario dm = new DemuxManagerScenario1(cc,2);  
      dm.start();
 
      List<String> requireActions = new ArrayList<String>();
     
      // DEMUX_INPUT_DIR reprocessing
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:true");
      requireActions.add("runDemux:true");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true");
      requireActions.add("processData done");

      // dataSink3.done processing
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:false"); 
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true"); 
      requireActions.add("runDemux:true"); 
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true"); 
      requireActions.add("processData done");
      
      List<String> actions = dm.actions;
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for (int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

    } catch (Exception e) {
      e.printStackTrace();
      Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
  }
  
  
  /**
   * DataSink file in dataSink directory
   * MR_INPUT_DIR already there
   * MR_OUTPUT_DIR already there
   * MR_OUTPUT_DIR should be deleted first
   * Then MR_INPUT_DIR should be reprocessed
   * Then dataSink file
   */
  public void testScenario4() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    cc.set(CHUKWA_CONSTANT.HDFS_DEFAULT_NAME_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );

    try {
      File mrInputDir = new File(chukwaRootDir + CHUKWA_CONSTANT.DEFAULT_DEMUX_PROCESSING_DIR_NAME+ CHUKWA_CONSTANT.DEFAULT_DEMUX_MR_INPUT_DIR_NAME);
      mrInputDir.mkdirs();
      File mrOutputDir = new File(chukwaRootDir + CHUKWA_CONSTANT.DEFAULT_DEMUX_PROCESSING_DIR_NAME+ CHUKWA_CONSTANT.DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME);
      mrOutputDir.mkdirs();
      
      File dataSinkDirectory = new File(chukwaRootDir +"/logs");
      dataSinkDirectory.mkdirs();
      File dataSinkFile = new File(chukwaRootDir +"/logs"+ "/dataSink4.done");
      dataSinkFile.createNewFile();
      
      DemuxManagerScenario dm = new DemuxManagerScenario1(cc,2);  
      dm.start();
 
      List<String> requireActions = new ArrayList<String>();

      requireActions.add("checkDemuxOutputDir:true"); 
      requireActions.add("deleteDemuxOutputDir:true");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:true");
      requireActions.add("runDemux:true"); 
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true"); 
      requireActions.add("processData done");

      // dataSink4.done processing
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:false"); 
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true"); 
      requireActions.add("runDemux:true"); 
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true"); 
      requireActions.add("processData done");
            
      List<String> actions = dm.actions;
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for(int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

    } catch (Exception e) {
      e.printStackTrace();
      Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
  }
  
  /**
   * DataSink file in dataSinkDir
   * Demux fails 3 times
   * Add a new DataSink file to dataSinkDir
   * Demux succeed
   */
  public void testScenario5() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    cc.set(CHUKWA_CONSTANT.HDFS_DEFAULT_NAME_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );

    try {
      File dataSinkDirectory = new File(chukwaRootDir +"/logs");
      dataSinkDirectory.mkdirs();
      File dataSinkFile = new File(chukwaRootDir +"/logs"+ "/dataSink5-0.done");
      dataSinkFile.createNewFile();
      
      DemuxManagerScenario dm = new DemuxManagerScenario5(cc);  
      dm.start();
 
      List<String> requireActions = new ArrayList<String>();
      
      // Move dataSink & process
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:false"); 
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true"); 
      requireActions.add("runDemux:false"); 
      requireActions.add("processData done"); 
      
      // Reprocess 1
      requireActions.add("checkDemuxOutputDir:true"); 
      requireActions.add("deleteDemuxOutputDir:true");
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:true"); 
      requireActions.add("runDemux:false"); 
      requireActions.add("processData done"); 
      
      // Reprocess 2
      requireActions.add("checkDemuxOutputDir:true"); 
      requireActions.add("deleteDemuxOutputDir:true");
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:true"); 
      requireActions.add("runDemux:false"); 
      requireActions.add("processData done"); 
      
      // Reprocess3
      requireActions.add("checkDemuxOutputDir:true"); 
      requireActions.add("deleteDemuxOutputDir:true");
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:true"); 
      requireActions.add("runDemux:false"); 
      requireActions.add("processData done"); 
      requireActions.add("checkDemuxOutputDir:true"); 
      requireActions.add("deleteDemuxOutputDir:true");
      requireActions.add("checkDemuxOutputDir:false");
      requireActions.add("checkDemuxInputDir:true");
      requireActions.add("moveDataSinkFilesToDemuxErrorDirectory:true");
     
      requireActions.add("checkDemuxOutputDir:false"); 
      requireActions.add("checkDemuxInputDir:false"); 
      requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true"); 
      requireActions.add("runDemux:true"); 
      requireActions.add("checkDemuxOutputDir:true");
      requireActions.add("moveDemuxOutputDirToPostProcessDirectory:true"); 
      requireActions.add("moveDataSinkFilesToArchiveDirectory:true"); 
      requireActions.add("processData done");
      
      
      List<String> actions = dm.actions;
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for(int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( i + " - " + requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

    } catch (Exception e) {
      e.printStackTrace();
      Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
  }
  
  
  /**
   * Standard workflow with MR_OUTPUT
   */
  public void testScenario6() {
    ChukwaConfiguration cc = new ChukwaConfiguration();
    String tempDirectory = System.getProperty("test.build.data", "/tmp");
    String chukwaRootDir = tempDirectory + "/demuxManagerTest_" + System.currentTimeMillis() +"/";
    
    
    cc.set(CHUKWA_CONSTANT.WRITER_HDFS_FILESYSTEM_FIELD, "file:///");
    cc.set(CHUKWA_CONSTANT.CHUKWA_ROOT_DIR_FIELD, chukwaRootDir );
    cc.set(CHUKWA_CONSTANT.CHUKWA_ARCHIVE_DIR_FIELD, chukwaRootDir +"/archives/" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_POST_PROCESS_DIR_FIELD, chukwaRootDir +"/postProcess" );
    cc.set(CHUKWA_CONSTANT.CHUKWA_DATA_SINK_DIR_FIELD, chukwaRootDir +"/logs" );
    
    try {
      
      File dataSinkDirectory = new File(chukwaRootDir +"/logs");
      dataSinkDirectory.mkdirs();
      File dataSinkFile = new File(chukwaRootDir +"/logs"+ "/dataSink6.done");
      dataSinkFile.createNewFile();
      
      DemuxManagerScenario dm = new DemuxManagerScenario6(cc,3);  
      dm.start();
      
      List<String> requireActions = new ArrayList<String>();
      for (int i=0;i<3;i++) {
        requireActions.add("checkDemuxOutputDir:false");
        requireActions.add("checkDemuxInputDir:false");
        requireActions.add("moveDataSinkFilesToDemuxInputDirectory:true");
        requireActions.add("runDemux:true");
        requireActions.add("checkDemuxOutputDir:true");
        requireActions.add("moveDemuxOutputDirToPostProcessDirectory:true");
        requireActions.add("moveDataSinkFilesToArchiveDirectory:true");
        requireActions.add("processData done");   
      }
     
      List<String> actions = dm.actions;
      Assert.assertTrue(requireActions.size() == actions.size());
      
      for(int i=0;i<requireActions.size();i++) {
        Assert.assertTrue( requireActions.get(i) + " == " +actions.get(i),requireActions.get(i).intern() == actions.get(i).intern());
      }

      
    } catch (Exception e) {
      e.printStackTrace();
     Assert.fail();
    }
    finally {
      deleteDirectory(new File(chukwaRootDir));
    }
    
  }
  
  
  static public boolean deleteDirectory(File path) {
    if( path.exists() ) {
      File[] files = path.listFiles();
      for(int i=0; i<files.length; i++) {
         if(files[i].isDirectory()) {
           deleteDirectory(files[i]);
         }
         else {
           files[i].delete();
         }
      }
    }
    return( path.delete() );
  }

     /////////////////////////\
    //// HELPER CLASSES /////  \
   /////////////////////////____\
  
  private static class DemuxManagerScenario6 extends DemuxManagerScenario {
    int count = 0;
    public DemuxManagerScenario6(ChukwaConfiguration conf, int count) throws Exception {
      super(conf);
      this.count = count;
    }
    @Override
    public boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      boolean res = super.runDemux(demuxInputDir, demuxOutputDir);
      
      try {
        // Create DEMUX_OUTPOUT
        String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD);
        File mrOutputDir = new File(chukwaRootDir  + DEFAULT_DEMUX_PROCESSING_DIR_NAME+ DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME);
        mrOutputDir.mkdirs();
        mrOutputDir.deleteOnExit();
        
        // ADD DATASINK FILE
        File dataSinkDirectory = new File(chukwaRootDir +"logs");
        dataSinkDirectory.mkdirs();
        File dataSinkFile = new File(chukwaRootDir +"logs"+ "/dataSink6-" + count + ".done");
        dataSinkFile.createNewFile();
        
      }catch(Exception e) {
        throw new RuntimeException(e);
      }

      count --;
      if (count <= 0) {
        this.isRunning = false;
      }
     
      return res;
    }
  }
  
  
  private static class DemuxManagerScenario5 extends DemuxManagerScenario {
    public DemuxManagerScenario5(ChukwaConfiguration conf) throws Exception {
      super(conf);
    }
    
    boolean errorDone = false;
    public boolean moveDataSinkFilesToDemuxErrorDirectory(String dataSinkDir,
        String demuxErrorDir) throws IOException {
      boolean res = super.moveDataSinkFilesToDemuxErrorDirectory(dataSinkDir, demuxErrorDir);
      
      String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD);
      File dataSinkDirectory = new File(chukwaRootDir +"logs");
      dataSinkDirectory.mkdirs();
      dataSinkDirectory.deleteOnExit();
      File dataSinkFile = new File(chukwaRootDir +"logs"+ "/dataSink5-1.done");
      dataSinkFile.createNewFile();
      dataSinkFile.deleteOnExit();       
  
      errorDone = true;
      return res;
    }
    
    int counter = 0;
    @Override
    public boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      if (errorDone && counter >= 4) {
        this.isRunning = false;
      } 
      
      // Create DEMUX_OUTPOUT
      String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD);
      File mrOutputDir = new File(chukwaRootDir  + DEFAULT_DEMUX_PROCESSING_DIR_NAME+ DEFAULT_DEMUX_MR_OUTPUT_DIR_NAME);
      mrOutputDir.mkdirs();
      mrOutputDir.deleteOnExit();
      
      counter ++;
      this.actions.add("runDemux:" + errorDone);
      return errorDone;
    }
  }
  
  private static class DemuxManagerScenario2 extends DemuxManagerScenario {
    public DemuxManagerScenario2(ChukwaConfiguration conf) throws Exception {
      super(conf);
    }
    
    int counter = 0;
    @Override
    public boolean moveDataSinkFilesToDemuxInputDirectory(String dataSinkDir,
        String demuxInputDir) throws IOException {
      if (counter == 1) {
        String chukwaRootDir = conf.get(CHUKWA_ROOT_DIR_FIELD);
        File dataSinkDirectory = new File(chukwaRootDir +"logs");
        dataSinkDirectory.mkdirs();
        dataSinkDirectory.deleteOnExit();
        File dataSinkFile = new File(chukwaRootDir +"logs"+ "/dataSink2.done");
        dataSinkFile.createNewFile();
        dataSinkFile.deleteOnExit();
      }

      counter ++;
      return super.moveDataSinkFilesToDemuxInputDirectory(dataSinkDir, demuxInputDir);
    }
    
   
    @Override
    public boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      boolean res = super.runDemux(demuxInputDir, demuxOutputDir);
      if (counter > 1) {
        this.isRunning = false;
      }
      return res;
    }
  }
  
  private static class DemuxManagerScenario1 extends DemuxManagerScenario {
    int count = 0;
    public DemuxManagerScenario1(ChukwaConfiguration conf, int count) throws Exception {
      super(conf);
      this.count = count;
    }
    @Override
    public boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      boolean res = super.runDemux(demuxInputDir, demuxOutputDir);
      count --;
      if (count <= 0) {
        this.isRunning = false;
      }
     
      return res;
    }
  }
  
  private static class DemuxManagerScenario extends DemuxManager {
    public List<String>actions = new ArrayList<String>();
 
    public DemuxManagerScenario(ChukwaConfiguration conf) throws Exception {
      super(conf);
      NO_DATASINK_SLEEP_TIME = 5;
    }

    @Override
    public boolean checkDemuxInputDir(String demuxInputDir) throws IOException {
      boolean res = super.checkDemuxInputDir(demuxInputDir);
      this.actions.add("checkDemuxInputDir:" + res);
      return res;
    }

    @Override
    public boolean checkDemuxOutputDir(String demuxOutputDir)
        throws IOException {
      boolean res = super.checkDemuxOutputDir(demuxOutputDir);
      this.actions.add("checkDemuxOutputDir:" + res);
      return res;
    }

    @Override
    public boolean moveDataSinkFilesToArchiveDirectory(String demuxInputDir,
        String archiveDirectory) throws IOException {
      boolean res = super.moveDataSinkFilesToArchiveDirectory(demuxInputDir,
          archiveDirectory);
      this.actions.add("moveDataSinkFilesToArchiveDirectory:" + res);
      return res;
    }

    @Override
    public boolean moveDataSinkFilesToDemuxErrorDirectory(String dataSinkDir,
        String demuxErrorDir) throws IOException {
      boolean res = super.moveDataSinkFilesToDemuxErrorDirectory(dataSinkDir, demuxErrorDir);
      this.actions.add("moveDataSinkFilesToDemuxErrorDirectory:" + res);
      return res;
    }

    @Override
    public boolean moveDataSinkFilesToDemuxInputDirectory(String dataSinkDir,
        String demuxInputDir) throws IOException {
      boolean res = super.moveDataSinkFilesToDemuxInputDirectory(dataSinkDir, demuxInputDir);
      this.actions.add("moveDataSinkFilesToDemuxInputDirectory:" + res);
      return res;
    }

    @Override
    public boolean moveDemuxOutputDirToPostProcessDirectory(
        String demuxOutputDir, String postProcessDirectory) throws IOException {
      boolean res = super.moveDemuxOutputDirToPostProcessDirectory(demuxOutputDir,
          postProcessDirectory);
      this.actions.add("moveDemuxOutputDirToPostProcessDirectory:" + res);
      return res;
    }

    @Override
    public boolean processData(String dataSinkDir, String demuxInputDir,
        String demuxOutputDir, String postProcessDir, String archiveDir)
        throws IOException {
      boolean res = super.processData(dataSinkDir, demuxInputDir, demuxOutputDir, postProcessDir,
          archiveDir);
      this.actions.add("processData done");
      return res;
    }

    @Override
    public boolean runDemux(String demuxInputDir, String demuxOutputDir) {
      boolean res = true;
      this.actions.add("runDemux:" + res);
      return res;
    }

    @Override
    public boolean deleteDemuxOutputDir(String demuxOutputDir)
        throws IOException {
      boolean res = super.deleteDemuxOutputDir(demuxOutputDir);
      this.actions.add("deleteDemuxOutputDir:" + res);
      return res;
    }
    
  }

}