Remove commons io usage and dependency. (#88)

* Remove commons io usage and dependency.

* Inverted while and try-with-resources surrounding the call to fileLoop.

Author: vjkoskela

pom.xml

@@ -79,7 +79,6 @@
     <aspectjrt.version>1.8.9</aspectjrt.version>
     <cglib.version>3.2.5</cglib.version>
     <client.protocol.version>0.10.0</client.protocol.version>
-    <commons.io.version>2.4</commons.io.version>
     <fastutil.version>8.1.1</fastutil.version>
     <findbugs.annotations.version>3.0.1</findbugs.annotations.version>
     <guava.version>23.4-jre</guava.version>
@@ -597,11 +596,6 @@
       <artifactId>fastutil</artifactId>
       <version>${fastutil.version}</version>
     </dependency>
-    <dependency>
-      <groupId>commons-io</groupId>
-      <artifactId>commons-io</artifactId>
-      <version>${commons.io.version}</version>
-    </dependency>
     <dependency>
       <groupId>org.reflections</groupId>
       <artifactId>reflections</artifactId>

src/main/java/com/arpnetworking/metrics/common/sources/FileSource.java

@@ -34,6 +34,7 @@
 import org.joda.time.Duration;
 import org.joda.time.Period;
 
+import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
@@ -67,6 +68,11 @@ public void stop() {
                     .setThrowable(e)
                     .log();
         }
+        try {
+            _positionStore.close();
+        } catch (final IOException e) {
+            // Ignore close exception.
+        }
     }
 
     /**
@@ -98,23 +104,23 @@ private FileSource(final Builder<T> builder) {
         super(builder);
         _logger = logger;
         _parser = builder._parser;
-        final PositionStore positionStore;
         if (builder._stateFile == null) {
-            positionStore = NO_POSITION_STORE;
+            _positionStore = NO_POSITION_STORE;
         } else {
-            positionStore = new FilePositionStore.Builder().setFile(builder._stateFile).build();
+            _positionStore = new FilePositionStore.Builder().setFile(builder._stateFile).build();
         }
 
         _tailer = new StatefulTailer.Builder()
                 .setFile(builder._sourceFile)
                 .setListener(new LogTailerListener())
                 .setReadInterval(builder._interval)
-                .setPositionStore(positionStore)
+                .setPositionStore(_positionStore)
                 .setInitialPosition(builder._initialPosition)
                 .build();
         _tailerExecutor = Executors.newSingleThreadExecutor(runnable -> new Thread(runnable, "FileSourceTailer"));
     }
 
+    private final PositionStore _positionStore;
     private final Parser<T, byte[]> _parser;
     private final Tailer _tailer;
     private final ExecutorService _tailerExecutor;

src/main/java/com/arpnetworking/metrics/common/tailer/StatefulTailer.java

@@ -24,7 +24,6 @@
 import com.arpnetworking.utility.Trigger;
 import com.google.common.base.MoreObjects;
 import net.sf.oval.constraint.NotNull;
-import org.apache.commons.io.IOUtils;
 import org.joda.time.Duration;
 
 import java.io.ByteArrayOutputStream;
@@ -68,11 +67,14 @@ public void run() {
                         .setThrowable(throwable)
                         .log());
 
-        try {
-            fileLoop();
-        } finally {
-            IOUtils.closeQuietly(_positionStore);
-            IOUtils.closeQuietly(_lineBuffer);
+        while (_isRunning) {
+            try (ByteArrayOutputStream lineBuffer = new ByteArrayOutputStream(INITIAL_BUFFER_SIZE)) {
+                _lineBuffer = lineBuffer;
+                fileLoop();
+            } catch (final IOException e) {
+                // Ignoring exception from closing line buffer because it's not a
+                // buffered output stream (e.g. nothing to flush).
+            }
         }
     }
 
@@ -107,23 +109,16 @@ protected boolean isRunning() {
     }
 
     private void fileLoop() {
-        SeekableByteChannel reader = null;
         InitialPosition nextInitialPosition = _initialPosition;
         try {
             while (isRunning()) {
                 // Attempt to open the file
-                try {
-                    reader = Files.newByteChannel(_file, StandardOpenOption.READ);
+                try (SeekableByteChannel reader = Files.newByteChannel(_file, StandardOpenOption.READ)) {
                     LOGGER.trace()
                             .setMessage("Opened file")
                             .addData("file", _file)
                             .log();
-                } catch (final NoSuchFileException e) {
-                    _listener.fileNotFound();
-                    _trigger.waitOnTrigger();
-                }
 
-                if (reader != null) {
                     // Position the reader
                     resume(reader, nextInitialPosition);
                     _listener.fileOpened();
@@ -135,9 +130,10 @@ private void fileLoop() {
                     readLoop(reader);
 
                     // Reset per file state
-                    IOUtils.closeQuietly(reader);
-                    reader = null;
                     _hash = Optional.empty();
+                } catch (final NoSuchFileException e) {
+                    _listener.fileNotFound();
+                    _trigger.waitOnTrigger();
                 }
             }
         // Clients may elect to kill the stateful tailer on an exception by calling stop, or they
@@ -153,8 +149,6 @@ private void fileLoop() {
             // CHECKSTYLE.ON: IllegalCatch
             handleThrowable(e);
         } finally {
-            IOUtils.closeQuietly(reader);
-            reader = null;
             _hash = Optional.empty();
         }
     }
@@ -515,7 +509,6 @@ private void handleThrowable(final Throwable t) {
         _trigger = trigger;
 
         _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
-        _lineBuffer = new ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
         try {
             _md5 = MessageDigest.getInstance("MD5");
         } catch (final NoSuchAlgorithmException e) {
@@ -536,14 +529,14 @@ private StatefulTailer(final Builder builder) {
     private final PositionStore _positionStore;
     private final TailerListener _listener;
     private final ByteBuffer _buffer;
-    private final ByteArrayOutputStream _lineBuffer;
     private final MessageDigest _md5;
     private final InitialPosition _initialPosition;
     private final Optional<Long> _maximumOffsetOnResume;
     private final Trigger _trigger;
 
     private volatile boolean _isRunning = true;
     private Optional<String> _hash = Optional.empty();
+    private ByteArrayOutputStream _lineBuffer;
 
     private static final int REQUIRED_BYTES_FOR_HASH = 512;
     private static final int INITIAL_BUFFER_SIZE = 65536;

src/test/java/com/arpnetworking/metrics/mad/performance/CollectdPipelinePT.java

@@ -19,8 +19,8 @@
 import com.carrotsearch.junitbenchmarks.BenchmarkOptions;
 import com.carrotsearch.junitbenchmarks.BenchmarkRule;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.io.ByteStreams;
 import com.google.common.io.Resources;
-import org.apache.commons.io.IOUtils;
 import org.joda.time.Duration;
 import org.junit.BeforeClass;
 import org.junit.Rule;
@@ -52,7 +52,7 @@ public static void setUp() throws IOException, URISyntaxException {
         final Path path = Paths.get("target/tmp/perf/collectd-sample1.log");
         final FileOutputStream outputStream = new FileOutputStream(path.toFile());
 
-        IOUtils.copy(gzipInputStream, outputStream);
+        ByteStreams.copy(gzipInputStream, outputStream);
 
         JSON_BENCHMARK_CONSUMER.prepareClass();
     }