[SERV-126] Allow audiowaveform processes to exit (#7)

* Make WaveformVerticle a worker verticle
* Fix IndexOutOfBoundsError when input CSV doesn't have IIIF Access URL

Author: Mark A. Matney, Jr

src/main/java/edu/ucla/library/avpairtree/Config.java

@@ -77,6 +77,11 @@ public final class Config {
      */
     public static final String CONVERSION_WORKERS = "conversion.workers";
 
+    /**
+     * The number of workers that should work to do audiowaveform generation.
+     */
+    public static final String WAVEFORM_WORKERS = "waveform.workers";
+
     /**
      * The environment variable for the S3 bucket for audio waveforms.
      */

src/main/java/edu/ucla/library/avpairtree/verticles/MainVerticle.java

@@ -132,7 +132,7 @@ private void configureServer(final JsonObject aConfig, final Promise<Void> aProm
                     futures.add(deployVerticle(new WatcherVerticle(), aConfig));
                     futures.add(deployVerticle(new PairtreeVerticle(), aConfig));
                     futures.add(deployVerticle(new ConverterVerticle(), aConfig.copy().put(WORKER, true)));
-                    futures.add(deployVerticle(new WaveformVerticle(), aConfig));
+                    futures.add(deployVerticle(new WaveformVerticle(), aConfig.copy().put(WORKER, true)));
 
                     CompositeFuture.all(futures).onSuccess(result -> {
                         startCsvDirWatcher(aConfig).onComplete(startup -> {
@@ -166,8 +166,17 @@ private Future<Void> deployVerticle(final Verticle aVerticle, final JsonObject a
 
         // If the configuration for this verticle mentions it should be a worker, find out how many to set
         if (aConfig.getBoolean(WORKER, false)) {
+            final int nWorkerThreads;
+
+            if (ConverterVerticle.class.equals(aVerticle.getClass())) {
+                nWorkerThreads = aConfig.getInteger(Config.CONVERSION_WORKERS, DEFAULT_WORKER_COUNT);
+            } else if (WaveformVerticle.class.equals(aVerticle.getClass())) {
+                nWorkerThreads = aConfig.getInteger(Config.WAVEFORM_WORKERS, DEFAULT_WORKER_COUNT);
+            } else {
+                nWorkerThreads = DEFAULT_WORKER_COUNT;
+            }
             options.setWorker(true).setWorkerPoolName(aVerticle.getClass().getSimpleName());
-            options.setWorkerPoolSize(aConfig.getInteger(Config.CONVERSION_WORKERS, DEFAULT_WORKER_COUNT));
+            options.setWorkerPoolSize(nWorkerThreads);
             options.setMaxWorkerExecuteTime(Integer.MAX_VALUE).setMaxWorkerExecuteTimeUnit(TimeUnit.MINUTES);
 
             LOGGER.debug(MessageCodes.AVPT_012, options.getWorkerPoolName(), options.getWorkerPoolSize());

src/main/java/edu/ucla/library/avpairtree/verticles/WatcherVerticle.java

@@ -200,8 +200,12 @@ private Future<String> updateCSV(final String aCsvFilePath, final Map<String, Cs
                             if (aCsvItemMap.containsKey(ark)) {
                                 row[index] = constructAccessURL(csvItem);
                             } else {
-                                // Don't overwrite what may already be there (e.g. in the case of images)
-                                row[index] = originalRow.get(index + 1);
+                                if (originalAccessUrlIndex != -1) {
+                                    // Don't overwrite what was already there (e.g. in the case of images)
+                                    row[index] = originalRow.get(index + 1);
+                                } else {
+                                    row[index] = "";
+                                }
                             }
                         } else if (waveformIndex == index) {
                             row[index] = aWaveformMap.getString(ark, "");

src/main/java/edu/ucla/library/avpairtree/verticles/WaveformVerticle.java

@@ -59,14 +59,14 @@ public void start(final Promise<Void> aPromise) {
         final JsonObject config = config();
         final String[] cmd = { "which", AUDIOWAVEFORM };
         final String cmdline = String.join(SPACE, cmd);
-        final ProcessBuilder pb = new ProcessBuilder(cmd).redirectOutput(ProcessBuilder.Redirect.PIPE);
-
         final String configErrorMsg;
 
+        LOGGER.debug(MessageCodes.AVPT_011, WaveformVerticle.class.getSimpleName(), Thread.currentThread().getName());
+
         // Make sure that audiowaveform is installed on the system
 
         try {
-            final Process which = pb.start();
+            final Process which = new ProcessBuilder(cmd).start();
             final int exitValue = which.waitFor();
             final InputStream stdout;
             final String cmdResult;
@@ -202,30 +202,28 @@ private Future<ByteBuffer> audiowaveform(final Path anAudioFilePath) throws IOEx
         final String[] cmd = { AUDIOWAVEFORM, "--input-filename", anAudioFilePath.toString(), "--output-format", "dat",
             "--bits", "8" };
         final String cmdline = String.join(SPACE, cmd);
-        final ProcessBuilder pb = new ProcessBuilder(cmd).redirectOutput(ProcessBuilder.Redirect.PIPE);
 
         try {
-            pb.start().onExit().thenAccept(subprocess -> {
-                try (InputStream stdout = subprocess.getInputStream();
-                        InputStream stderr = subprocess.getErrorStream()) {
-
-                    final int exitValue = subprocess.exitValue();
+            final Process audiowaveform = new ProcessBuilder(cmd).start();
 
-                    if (0 == exitValue) {
-                        // Redact the binary audiowaveform data for logging
-                        final String cmdResultMsg = LOGGER.getMessage(MessageCodes.AVPT_015, cmdline, exitValue,
-                                "[binary audiowaveform data]");
+            // Unless we read its output before calling `onExit()`, the audiowaveform process will stay asleep until it
+            // receives an interrupt signal
+            final byte[] stdout = audiowaveform.getInputStream().readAllBytes();
+            final String stderr = new String(audiowaveform.getErrorStream().readAllBytes(), StandardCharsets.UTF_8);
 
-                        LOGGER.debug(cmdResultMsg);
+            audiowaveform.onExit().thenAccept(process -> {
+                final int exitValue = process.exitValue();
 
-                        asyncResult.complete(ByteBuffer.wrap(stdout.readAllBytes()));
-                    } else {
-                        final String errorOutput = new String(stderr.readAllBytes());
-
-                        asyncResult.fail(LOGGER.getMessage(MessageCodes.AVPT_015, cmdline, exitValue, errorOutput));
+                if (0 == exitValue) {
+                    for (final String line : stderr.split("\\r?\\n")) {
+                        LOGGER.debug(line);
                     }
-                } catch (final IOException details) {
-                    asyncResult.fail(LOGGER.getMessage(MessageCodes.AVPT_016, cmdline, details));
+                    // Redact the binary audiowaveform data for logging
+                    LOGGER.debug(MessageCodes.AVPT_015, cmdline, exitValue, "[binary audiowaveform data]");
+
+                    asyncResult.complete(ByteBuffer.wrap(stdout));
+                } else {
+                    asyncResult.fail(LOGGER.getMessage(MessageCodes.AVPT_015, cmdline, exitValue, stderr));
                 }
             });
         } catch (final IOException details) {

src/test/resources/test-config.properties

@@ -28,3 +28,6 @@ iiif.access.url.id.index = 1
 
 # The number of threads working on media file conversions
 conversion.workers = 2
+
+# The number of threads working on audiowaveform generation
+waveform.workers = 2