Refactor progress parsing to handle exceptions gracefully

src/main/java/net/bramp/ffmpeg/FFmpegUtils.java

@@ -107,15 +107,11 @@ public static String millisToSeconds(long millis) {
    * format "hour:minute:second", where second can be a decimal number.
    *
    * @param time the timecode to parse.
-   * @return the number of nanoseconds or -1 if time is 'N/A'
+   * @return the number of nanoseconds
    */
   public static long fromTimecode(String time) {
     checkNotEmpty(time, "time must not be empty string");
 
-    if (time.equals("N/A")) {
-      return -1;
-    }
-
     Matcher m = TIME_REGEX.matcher(time);
     if (!m.find()) {
       throw new IllegalArgumentException("invalid time '" + time + "'");
@@ -134,14 +130,11 @@ public static long fromTimecode(String time) {
    * Converts a string representation of bitrate to a long of bits per second.
    *
    * @param bitrate in the form of 12.3kbits/s
-   * @return the bitrate in bits per second or -1 if bitrate is 'N/A'
+   * @return the bitrate in bits per second
    */
   public static long parseBitrate(String bitrate) {
     checkNotEmpty(bitrate, "bitrate must not be empty string");
 
-    if ("N/A".equals(bitrate)) {
-      return -1;
-    }
     Matcher m = BITRATE_REGEX.matcher(bitrate);
     if (!m.find()) {
       throw new IllegalArgumentException("Invalid bitrate '" + bitrate + "'");

src/main/java/net/bramp/ffmpeg/progress/Progress.java

@@ -14,7 +14,7 @@
 // TODO: Change to be immutable
 /** Represents progress data reported by FFmpeg during encoding. */
 public class Progress {
-  static final Logger LOG = LoggerFactory.getLogger(Progress.class);
+  static final Logger logger = LoggerFactory.getLogger(Progress.class);
 
   /** Enum representing the status of FFmpeg progress updates. */
   public enum Status {
@@ -85,7 +85,7 @@ public Progress() {
   /** Constructs a progress instance with the specified values. */
   public Progress(
       long frame,
-      float fps,
+      Fraction fps,
       long bitrate,
       long total_size,
       long out_time_ns,
@@ -94,7 +94,7 @@ public Progress(
       float speed,
       Status status) {
     this.frame = frame;
-    this.fps = Fraction.getFraction(fps);
+    this.fps = fps;
     this.bitrate = bitrate;
     this.total_size = total_size;
     this.out_time_ns = out_time_ns;
@@ -108,7 +108,7 @@ public Progress(
    * Parses values from the line, into this object.
    *
    * <p>The value options are defined in ffmpeg.c's print_report function
-   * https://github.com/FFmpeg/FFmpeg/blob/master/ffmpeg.c
+   * https://github.com/FFmpeg/FFmpeg/blob/master/fftools/ffmpeg.c
    *
    * @param line A single line of output from ffmpeg
    * @return true if the record is finished
@@ -130,26 +130,38 @@ protected boolean parseLine(String line) {
 
     switch (key) {
       case "frame":
-        frame = Long.parseLong(value);
+        try {
+          frame = Long.parseLong(value);
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse frame: {}", value);
+          frame = -1;
+        }
         return false;
 
       case "fps":
-        fps = Fraction.getFraction(value);
+        try {
+          fps = Fraction.getFraction(value);
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse fps: {}", value);
+          fps = null;
+        }
         return false;
 
       case "bitrate":
-        if (value.equals("N/A")) {
-          bitrate = -1;
-        } else {
+        try {
           bitrate = FFmpegUtils.parseBitrate(value);
+        } catch (IllegalArgumentException e) {
+          logger.warn("Failed to parse bitrate: {}", value);
+          bitrate = -1;
         }
         return false;
 
       case "total_size":
-        if (value.equals("N/A")) {
-          total_size = -1;
-        } else {
+        try {
           total_size = Long.parseLong(value);
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse total_size: {}", value);
+          total_size = -1;
         }
         return false;
 
@@ -164,28 +176,48 @@ protected boolean parseLine(String line) {
         return false;
 
       case "out_time":
-        out_time_ns = fromTimecode(value);
+        try {
+          out_time_ns = fromTimecode(value);
+        } catch (IllegalArgumentException e) {
+          logger.warn("Failed to parse out_time: {}", value);
+          out_time_ns = -1;
+        }
         return false;
 
       case "dup_frames":
-        dup_frames = Long.parseLong(value);
+        try {
+          dup_frames = Long.parseLong(value);
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse dup_frames: {}", value);
+          dup_frames = -1;
+        }
         return false;
 
       case "drop_frames":
-        drop_frames = Long.parseLong(value);
+        try {
+          drop_frames = Long.parseLong(value);
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse drop_frames: {}", value);
+          drop_frames = -1;
+        }
         return false;
 
       case "speed":
-        if (value.equals("N/A")) {
-          speed = -1;
-        } else {
+        try {
           speed = Float.parseFloat(value.replace("x", ""));
+        } catch (NumberFormatException e) {
+          logger.warn("Failed to parse speed: {}", value);
+          speed = -1;
         }
         return false;
 
       case "progress":
         // TODO: After "end" stream is closed
-        status = Status.of(value);
+        try {
+          status = Status.of(value);
+        } catch (IllegalArgumentException e) {
+          logger.warn("Failed to parse progress status: {}", value);
+        }
         return true; // The status field is always last in the record
 
       default:
@@ -196,7 +228,7 @@ protected boolean parseLine(String line) {
           // AV_CODEC_FLAG_PSNR
           // stream_%d_%d_psnr_all
         } else {
-          LOG.warn("skipping unhandled key: {} = {}", key, value);
+          logger.warn("Skipping unhandled key: {} = {}", key, value);
         }
 
         return false; // Either way, not supported

src/test/java/net/bramp/ffmpeg/FFmpegUtilsTest.java

@@ -61,7 +61,6 @@ public void testParseBitrate() {
     assertEquals(12300, parseBitrate("12.3kbits/s"));
     assertEquals(1000, parseBitrate("1kbits/s"));
     assertEquals(123, parseBitrate("0.123kbits/s"));
-    assertEquals(-1, parseBitrate("N/A"));
   }
 
   @Test(expected = IllegalArgumentException.class)

src/test/java/net/bramp/ffmpeg/fixtures/Progresses.java

@@ -2,6 +2,7 @@
 
 import com.google.common.collect.ImmutableList;
 import net.bramp.ffmpeg.progress.Progress;
+import org.apache.commons.lang3.math.Fraction;
 
 public final class Progresses {
 
@@ -16,11 +17,44 @@ private Progresses() {
 
   public static final ImmutableList<Progress> allProgresses =
       ImmutableList.of(
-          new Progress(5, 0.0f, 800, 48, 512000000, 0, 0, 1.01f, Progress.Status.CONTINUE),
-          new Progress(118, 23.4f, -1, -1, 5034667000L, 0, 0, -1, Progress.Status.CONTINUE),
           new Progress(
-              132, 23.1f, 1935500, 1285168, 5312000000L, 0, 0, 0.929f, Progress.Status.END));
+              5,
+              Fraction.getFraction(0.0f),
+              800,
+              48,
+              512000000,
+              0,
+              0,
+              1.01f,
+              Progress.Status.CONTINUE),
+          new Progress(
+              118,
+              Fraction.getFraction(23.4f),
+              -1,
+              -1,
+              5034667000L,
+              0,
+              0,
+              -1,
+              Progress.Status.CONTINUE),
+          new Progress(
+              132,
+              Fraction.getFraction(23.1f),
+              1935500,
+              1285168,
+              5312000000L,
+              0,
+              0,
+              0.929f,
+              Progress.Status.END));
 
   public static final ImmutableList<Progress> naProgresses =
-      ImmutableList.of(new Progress(0, 0.0f, -1, -1, -1, 0, 0, -1, Progress.Status.END));
+      ImmutableList.of(
+          new Progress(0, Fraction.getFraction(0.0f), -1, -1, -1, 0, 0, -1, Progress.Status.END));
+
+  public static final ImmutableList<String> malformedProgressFile =
+      ImmutableList.of("ffmpeg-progress-malformed");
+
+  public static final ImmutableList<Progress> malformedProgresses =
+      ImmutableList.of(new Progress(-1, null, 1935500, -1, 522000000000L, -1, -1, 0.929f, null));
 }

src/test/java/net/bramp/ffmpeg/progress/StreamProgressParserTest.java

@@ -36,4 +36,16 @@ public void testNaProgressPackets() throws IOException {
 
     assertThat(listener.progesses, equalTo(Progresses.naProgresses));
   }
+
+  @Test
+  public void testMalformedProgressPackets() throws IOException {
+    listener.reset();
+
+    StreamProgressParser parser = new StreamProgressParser(listener);
+
+    InputStream inputStream = combineResource(Progresses.malformedProgressFile);
+    parser.processStream(inputStream);
+
+    assertThat(listener.progesses, equalTo(Progresses.malformedProgresses));
+  }
 }

src/test/java/net/bramp/ffmpeg/progress/TcpProgressParserTest.java

@@ -24,7 +24,7 @@ public ProgressParser newParser(ProgressListener listener)
   }
 
   @Test
-  public void testNormal() throws IOException, InterruptedException, URISyntaxException {
+  public void testNormal() throws IOException, InterruptedException {
     parser.start();
 
     Socket client = new Socket(uri.getHost(), uri.getPort());
@@ -47,7 +47,7 @@ public void testNormal() throws IOException, InterruptedException, URISyntaxExce
   }
 
   @Test
-  public void testNaProgressPackets() throws IOException, InterruptedException, URISyntaxException {
+  public void testNaProgressPackets() throws IOException, InterruptedException {
     parser.start();
 
     Socket client = new Socket(uri.getHost(), uri.getPort());
@@ -69,6 +69,29 @@ public void testNaProgressPackets() throws IOException, InterruptedException, UR
     assertThat(progesses, equalTo(Progresses.naProgresses));
   }
 
+  @Test
+  public void testMalformedProgressPackets() throws IOException, InterruptedException {
+    parser.start();
+
+    Socket client = new Socket(uri.getHost(), uri.getPort());
+    assertTrue("Socket is connected", client.isConnected());
+
+    InputStream inputStream = combineResource(Progresses.malformedProgressFile);
+    OutputStream outputStream = client.getOutputStream();
+
+    long bytes = ByteStreams.copy(inputStream, outputStream);
+
+    // HACK, but give the TcpProgressParser thread time to actually handle the connection/data
+    // before the client is closed, and the parser is stopped.
+    Thread.sleep(100);
+
+    client.close();
+    parser.stop();
+
+    assertThat(bytes, greaterThan(0L));
+    assertThat(progesses, equalTo(Progresses.malformedProgresses));
+  }
+
   @Test
   public void testPrematureDisconnect() throws IOException {
     parser.start();

src/test/java/net/bramp/ffmpeg/progress/UdpProgressParserTest.java

@@ -48,7 +48,7 @@ public void testNormal() throws IOException, InterruptedException {
   }
 
   @Test
-  public void testNaProgressPackets() throws IOException, InterruptedException, URISyntaxException {
+  public void testNaProgressPackets() throws IOException, InterruptedException {
     parser.start();
 
     final InetAddress addr = InetAddress.getByName(uri.getHost());
@@ -71,4 +71,29 @@ public void testNaProgressPackets() throws IOException, InterruptedException, UR
 
     assertThat(progesses, equalTo(Progresses.naProgresses));
   }
+
+  @Test
+  public void testMalformedProgressPackets() throws IOException, InterruptedException {
+    parser.start();
+
+    final InetAddress addr = InetAddress.getByName(uri.getHost());
+    final int port = uri.getPort();
+
+    try (DatagramSocket socket = new DatagramSocket()) {
+      // Load each Progress Fixture, and send in a single datagram packet
+      for (String progressFixture : Progresses.malformedProgressFile) {
+        InputStream inputStream = loadResource(progressFixture);
+        byte[] bytes = ByteStreams.toByteArray(inputStream);
+
+        DatagramPacket packet = new DatagramPacket(bytes, bytes.length, addr, port);
+        socket.send(packet);
+      }
+    }
+
+    Thread.sleep(100); // HACK: Wait a short while to avoid closing the receiving socket
+
+    parser.stop();
+
+    assertThat(progesses, equalTo(Progresses.malformedProgresses));
+  }
 }

src/test/resources/net/bramp/ffmpeg/fixtures/ffmpeg-progress-malformed

@@ -0,0 +1,11 @@
+frame=N/A
+fps=N/A
+stream_0_0_q=-1.0
+bitrate=1935.5kbits/s
+total_size=0.1
+out_time_ms=N/A
+out_time=00:00:0522
+dup_frames=N/A
+drop_frames=N/A
+speed=0.929x
+progress=err