package com.github.tstevens.storm.hdfs.spout;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.EnumSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.inotify.Event;
import org.hamcrest.CoreMatchers;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.testing.TupleCaptureBolt;
import backtype.storm.topology.TopologyBuilder;

public class HdfsSpoutTest {

    @ClassRule
    public static TemporaryFolder folder = new TemporaryFolder();

    private static MiniDFSCluster hdfsCluster;

    private static String hdfsURI;

    @BeforeClass
    public static void beforeClass() throws IOException{
        File baseDir = folder.newFolder("hdfs").getAbsoluteFile();

        Configuration conf = new Configuration(true);

        conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, baseDir.getAbsolutePath());
        MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(conf);

        hdfsCluster = builder.build();
        hdfsCluster.waitActive();

        hdfsURI = "hdfs://localhost:"+ hdfsCluster.getNameNodePort() + "/";
    }

    @AfterClass
    public static void afterClass(){
        if(hdfsCluster != null){
            hdfsCluster.shutdown();
        }
        hdfsURI = null;
    }

    @Test
    public void testSpout(){
    	URI hdfsUri;
		try {
            hdfsUri = new URI(hdfsURI);
		} catch (URISyntaxException e1) {
			e1.printStackTrace();
			Assert.fail();
			return;
		}

        HdfsInotifySpout spout = new HdfsInotifySpout(hdfsUri, "/", EnumSet.of(Event.EventType.CLOSE));

        TupleCaptureBolt capture = new TupleCaptureBolt();

        Config conf = new Config();

        TopologyBuilder builder = new TopologyBuilder();

        builder.setSpout("spout", spout, 1).setNumTasks(1);
        builder.setBolt("bolt", capture, 1).setNumTasks(1).shuffleGrouping("spout", HdfsInotifySpout.STREAM_ID);

        LocalCluster cluster = new LocalCluster();

        cluster.submitTopology("test-topo", conf, builder.createTopology());

        try {
            Thread.sleep(2500);

            hdfsCluster.getFileSystem().createNewFile(new Path("/test.txt"));

            Thread.sleep(200);
        } catch (IllegalArgumentException | IOException | InterruptedException e) {
            Assert.fail(e.getMessage());
        }

        cluster.shutdown();

        Assert.assertThat(capture.getResults().size(), CoreMatchers.is(1));
    }
}

@@ -899,22 +899,19 @@
 tic 
-String
+URI
  hdfsU
-RI
+ri
 ;%0A%0A 
@@ -1334,24 +1334,42 @@
 tActive();%0A%0A
+        try %7B%0A    
         hdfs
@@ -1369,21 +1369,29 @@
    hdfsU
-RI
+ri
  = 
+new URI(
 %22hdfs://
@@ -1438,18 +1438,152 @@
 () + %22/%22
+)
 ;%0A
+        %7D catch (URISyntaxException e1) %7B%0A            e1.printStackTrace();%0A            Assert.fail();%0A            return;%0A        %7D%0A
     %7D%0A%0A 
@@ -1726,18 +1726,18 @@
    hdfsU
-RI
+ri
  = null;
@@ -1787,169 +1787,8 @@
 ()%7B%0A
-    %09URI hdfsUri;%0A%09%09try %7B%0A            hdfsUri = new URI(hdfsURI);%0A%09%09%7D catch (URISyntaxException e1) %7B%0A%09%09%09e1.printStackTrace();%0A%09%09%09Assert.fail();%0A%09%09%09return;%0A%09%09%7D%0A%0A
     

Generate HDFS uri within BeforeClass