@Test
@Ignore("Does not work on Windows")
public void testHdfsSink() throws IOException {
    File destDir = temp.newFolder("hdfs");

    TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
    runner.setProperty(ExecuteFlumeSink.SINK_TYPE, "hdfs");
    runner.setProperty(ExecuteFlumeSink.FLUME_CONFIG,
        "tier1.sinks.sink-1.hdfs.path = " + destDir.toURI().toString() + "\n" +
        "tier1.sinks.sink-1.hdfs.fileType = DataStream\n" +
        "tier1.sinks.sink-1.hdfs.serializer = TEXT\n" +
        "tier1.sinks.sink-1.serializer.appendNewline = false"
    );
    try (InputStream inputStream = getClass().getResourceAsStream("/testdata/records.txt")) {
        Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), "records.txt");
        runner.enqueue(inputStream, attributes);
        runner.run();
    }

    File[] files = destDir.listFiles((FilenameFilter)HiddenFileFilter.VISIBLE);
    assertEquals("Unexpected number of destination files.", 1, files.length);
    File dst = files[0];
    byte[] expectedMd5;
    try (InputStream md5Stream = getClass().getResourceAsStream("/testdata/records.txt")) {
        expectedMd5 = FileUtils.computeMd5Digest(md5Stream);
    }
    byte[] actualMd5 = FileUtils.computeMd5Digest(dst);
    Assert.assertArrayEquals("Destination file doesn't match source data", expectedMd5, actualMd5);
}
 

/**
 * This method is supposed to be called from subclass to initialize data loading process.<br>
 * <br>
 * This method is using file given in the constructor to load the data and there are two possibilities:
 * <ul>
 * <li>Given file is file is in deed the <b>file</b> then it's forwarded to {@link #loadFile(File)} method</li>
 * <li>Given file is a <b>directory</b>, then this method is obtaining list of all visible .txt files in this directory and subdirectiores ( except hidden ones and those named "new" ) and call
 * {@link #loadFile(File)} for each of these files.
 * </ul>
 */
@SuppressWarnings("deprecation")
protected void loadData() {
	if (dataFile.isDirectory()) {
		Collection<?> files = FileUtils.listFiles(dataFile, FileFilterUtils.andFileFilter(FileFilterUtils.andFileFilter(FileFilterUtils.notFileFilter(FileFilterUtils.nameFileFilter("new")), FileFilterUtils.suffixFileFilter(".txt")), HiddenFileFilter.VISIBLE), HiddenFileFilter.VISIBLE);

		for (Object file1 : files) {
			File f = (File) file1;
			loadFile(f);
		}
	}
	else {
		loadFile(dataFile);
	}
}
 

@Test
@Ignore("Does not work on Windows")
public void testHdfsSink() throws IOException {
    File destDir = temp.newFolder("hdfs");

    TestRunner runner = TestRunners.newTestRunner(ExecuteFlumeSink.class);
    runner.setProperty(ExecuteFlumeSink.SINK_TYPE, "hdfs");
    runner.setProperty(ExecuteFlumeSink.FLUME_CONFIG,
        "tier1.sinks.sink-1.hdfs.path = " + destDir.toURI().toString() + "\n" +
        "tier1.sinks.sink-1.hdfs.fileType = DataStream\n" +
        "tier1.sinks.sink-1.hdfs.serializer = TEXT\n" +
        "tier1.sinks.sink-1.serializer.appendNewline = false"
    );
    try (InputStream inputStream = getClass().getResourceAsStream("/testdata/records.txt")) {
        Map<String, String> attributes = new HashMap<>();
        attributes.put(CoreAttributes.FILENAME.key(), "records.txt");
        runner.enqueue(inputStream, attributes);
        runner.run();
    }

    File[] files = destDir.listFiles((FilenameFilter)HiddenFileFilter.VISIBLE);
    assertEquals("Unexpected number of destination files.", 1, files.length);
    File dst = files[0];
    byte[] expectedMd5;
    try (InputStream md5Stream = getClass().getResourceAsStream("/testdata/records.txt")) {
        expectedMd5 = FileUtils.computeMd5Digest(md5Stream);
    }
    byte[] actualMd5 = FileUtils.computeMd5Digest(dst);
    Assert.assertArrayEquals("Destination file doesn't match source data", expectedMd5, actualMd5);
}
 

public List<String> getFolders()
    throws MojoExecutionException
{
    IOFileFilter folders = FileFilterUtils.and( HiddenFileFilter.VISIBLE, DirectoryFileFilter.DIRECTORY );
    IOFileFilter makeSVNAware = FileFilterUtils.makeSVNAware( folders );
    IOFileFilter makeCVSAware = FileFilterUtils.makeCVSAware( makeSVNAware );

    String[] list = folder.list( makeCVSAware );
    if ( list == null )
    {
        throw new MojoExecutionException( "The specified folder doesn't exist: " + folder );
    }

    List<File> listOfDirectories = new ArrayList<File>();
    for ( String item : list )
    {
        listOfDirectories.add( new File( folder, item ) );
    }

    Collections.sort( listOfDirectories, convertSortOrder() );
    List<String> resultList = new ArrayList<String>();
    for ( File file : listOfDirectories )
    {
        resultList.add( file.getName() );
    }
    return resultList;
}
 

private Collection<File> listFiles(File root, boolean recursive) {
	IOFileFilter dirFilter = recursive ? makeSVNAware(HiddenFileFilter.VISIBLE) : null;

	return FileUtils.listFiles(root, and(and(notFileFilter(prefixFileFilter("new")), suffixFileFilter(".xml")), HiddenFileFilter.VISIBLE), dirFilter);
}
 

protected Collection<File> listFiles(File root, boolean recursive) {
	IOFileFilter dirFilter = recursive ? makeSVNAware(HiddenFileFilter.VISIBLE) : null;
	return FileUtils.listFiles(root, and(and(notFileFilter(prefixFileFilter("new")), suffixFileFilter(".xml")), HiddenFileFilter.VISIBLE), dirFilter);
}
 

@SuppressWarnings("deprecation")
private static Collection<File> listFiles(File root, boolean recursive) {
	IOFileFilter dirFilter = recursive ? makeSVNAware(HiddenFileFilter.VISIBLE) : null;

	return FileUtils.listFiles(root, andFileFilter(andFileFilter(notFileFilter(prefixFileFilter("new")), suffixFileFilter(".xml")), HiddenFileFilter.VISIBLE), dirFilter);
}
 

public LicensableFileDirectoryWalker( IOFileFilter fileFilter, String firstLine, String linePrefix, String lastLine ) {
    super(HiddenFileFilter.VISIBLE,fileFilter,100);
    this.firstLine = firstLine;
    this.linePrefix = linePrefix;
    this.lastLine = lastLine;
}
 

/**
 * @param args
 * @throws Exception
 */
public static void main(String[] args) throws Exception {


    // 监控根目录，会递归监控子文件夹变化

    String rootDir = "D:\\download";
    // 轮询间隔 5 秒，通过重新启动 FileAlterationMonitor 实现
    long interval = TimeUnit.SECONDS.toMillis(5);


    // 创建一个文件观察器用于处理文件的格式
    // Create a FileFilter
    IOFileFilter directories = FileFilterUtils.and(FileFilterUtils.directoryFileFilter(), HiddenFileFilter.VISIBLE);
    IOFileFilter files = FileFilterUtils.and(FileFilterUtils.fileFileFilter(), FileFilterUtils.suffixFileFilter(".java"));
    IOFileFilter filter = FileFilterUtils.or(directories, files);

    // Create the File system observer and register File Listeners
    FileAlterationObserver observer = new FileAlterationObserver(rootDir, filter, null);
    observer.addListener(new FileMonitorFileListener()); //设置文件变化监听器

    //创建文件变化监听器
    FileAlterationMonitor monitor = new FileAlterationMonitor(interval, observer);

    // 开始监控
    monitor.start();

}
 

/**
 * 获取文件过滤器
 *
 * @return
 */
public static FileFilter getFileFilter() {
    return HiddenFileFilter.VISIBLE;
}