Progress bar shows actual progress + fix for empty cells + menu fix.

Author: Robin Duda

src/main/java/com/codingchili/ApplicationLauncher.java

@@ -27,7 +27,7 @@
  */
 public class ApplicationLauncher {
     private static final Logger logger = Logger.getLogger(ApplicationLauncher.class.getName());
-    public static String version = "1.2.2";
+    public static String VERSION = "1.2.3";
     private Vertx vertx = Vertx.vertx();
     private String[] args;
 
@@ -38,7 +38,7 @@ public static void main(String[] args) {
     public ApplicationLauncher(String[] args) {
         this.args = args;
 
-        logger.info(String.format("Starting excelastic %s..", version));
+        logger.info(String.format("Starting excelastic %s..", VERSION));
         logger.info("to import files without the web interface use: <filename> <index> <mapping>");
         Future<Void> future = Future.future();
 

src/main/java/com/codingchili/Controller/Atomic.java

@@ -0,0 +1,17 @@
+package com.codingchili.Controller;
+
+public class Atomic<E> {
+    private E reference;
+
+    public Atomic(E reference) {
+        this.reference = reference;
+    }
+
+    public E get() {
+        return reference;
+    }
+
+    public void set(E reference) {
+        this.reference = reference;
+    }
+}

src/main/java/com/codingchili/Controller/Website.java

@@ -1,36 +1,44 @@
 package com.codingchili.Controller;
 
-import com.codingchili.ApplicationLauncher;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.Iterator;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
 import com.codingchili.Model.Configuration;
 import com.codingchili.Model.ElasticWriter;
 import com.codingchili.Model.FileParser;
 import com.codingchili.Model.ParserException;
-import io.vertx.core.*;
+import static com.codingchili.ApplicationLauncher.VERSION;
+import static com.codingchili.Model.Configuration.INDEXING_ELASTICSEARCH;
+import static com.codingchili.Model.ElasticWriter.*;
+import static com.codingchili.Model.FileParser.INDEX;
+
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Context;
+import io.vertx.core.Future;
+import io.vertx.core.MultiMap;
+import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.eventbus.DeliveryOptions;
+import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.FileUpload;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.handler.BodyHandler;
 import io.vertx.ext.web.handler.StaticHandler;
 import io.vertx.ext.web.handler.TemplateHandler;
 import io.vertx.ext.web.templ.JadeTemplateEngine;
 
-import java.io.PrintWriter;
-import java.io.StringWriter;
-import java.util.Iterator;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import static com.codingchili.Model.ElasticWriter.INDEXING_TIMEOUT;
-import static com.codingchili.Model.FileParser.INDEX;
-
 /**
  * @author Robin Duda
- *         <p>
- *         Manages the web interface and handles file uploads.
+ * <p>
+ * Manages the web interface and handles file uploads.
  */
 public class Website extends AbstractVerticle {
     public static final String MAPPING = "mapping";
+    public static final String UPLOAD_ID = "uploadId";
     private Logger logger = Logger.getLogger(getClass().getName());
     private static final String DONE = "/done";
     private static final String ERROR = "/error";
@@ -57,7 +65,7 @@ public void start(Future<Void> start) {
 
         // adds values used in the template to all routes.
         router.route("/*").handler(context -> {
-            context.put("version", ApplicationLauncher.version);
+            context.put("version", VERSION);
             context.put("esVersion", ElasticWriter.getElasticVersion());
             context.put("esURL", Configuration.getElasticURL());
             context.put("connected", ElasticWriter.isConnected());
@@ -66,7 +74,25 @@ public void start(Future<Void> start) {
 
         router.route("/*").handler(TemplateHandler.create(JadeTemplateEngine.create()));
 
-        vertx.createHttpServer().requestHandler(router::accept).listen(Configuration.getWebPort(), done -> {
+        vertx.createHttpServer().websocketHandler(websock -> {
+            websock.writeFinalTextFrame(new JsonObject().put("message", "websocket connected to excelastic " + VERSION +
+                " using ElasticSearch " + getElasticVersion()).encode());
+
+            Atomic<String> uploadId = new Atomic<>("");
+            MessageConsumer<JsonObject> consumer = vertx.eventBus().consumer(IMPORT_PROGRESS, data -> {
+                try {
+                    if (uploadId.get().equals(data.body().getString(UPLOAD_ID))) {
+                        websock.writeFinalTextFrame(data.body().encode());
+                    }
+                } catch (Throwable e) {
+                    websock.close();
+                }
+            });
+            websock.handler(handler -> uploadId.set(handler.toJsonObject().getString(UPLOAD_ID)));
+            websock.closeHandler(closed -> consumer.unregister());
+            websock.exceptionHandler(sock -> consumer.unregister());
+
+        }).requestHandler(router::accept).listen(Configuration.getWebPort(), done -> {
             if (done.succeeded()) {
                 Configuration.setWebPort(done.result().actualPort());
                 logger.info("Started website on port " + Configuration.getWebPort());
@@ -79,35 +105,37 @@ public void start(Future<Void> start) {
 
     /**
      * Adds the upload route to the given router
+     *
      * @param router the upload route is added to the given router
      */
     private void setRouterAPI(Router router) {
         router.route("/api/upload").handler(context -> {
             Iterator<FileUpload> iterator = context.fileUploads().iterator();
 
             if (iterator.hasNext()) {
-                logger.info("Receiving uploaded file.. ");
+                MultiMap params = context.request().params();
+                logger.info("Receiving uploaded file with request id " + params.get(UPLOAD_ID));
                 FileUpload upload = context.fileUploads().iterator().next();
 
                 vertx.fileSystem().readFile(upload.uploadedFileName(), file -> {
-                    parse(file.result(), context.request().params(), upload.fileName(),
-                            Future.<Integer>future().setHandler(result -> {
-                                if (result.succeeded()) {
-                                    String index = context.request().params().get(INDEX);
-                                    logger.info(String.format("Imported file '%s' successfully into '%s'.",
-                                            upload.fileName(), index));
-
-                                    context.put(INDEX, index);
-                                    context.put(FILE, upload.fileName());
-                                    context.put(IMPORTED, result.result());
-                                    context.reroute(DONE);
-                                } else {
-                                    context.put(MESSAGE, traceToText(result.cause()));
-                                    logger.log(Level.SEVERE, String.format("Failed to parse file '%s'.",
-                                            upload.fileName()), result.cause());
-                                    context.reroute(ERROR);
-                                }
-                            }));
+                    parse(file.result(), params, upload.fileName(),
+                        Future.<Integer>future().setHandler(result -> {
+                            if (result.succeeded()) {
+                                String index = context.request().params().get(INDEX);
+                                logger.info(String.format("Imported file '%s' successfully into '%s'.",
+                                    upload.fileName(), index));
+
+                                context.put(INDEX, index);
+                                context.put(FILE, upload.fileName());
+                                context.put(IMPORTED, result.result());
+                                context.reroute(DONE);
+                            } else {
+                                context.put(MESSAGE, traceToText(result.cause()));
+                                logger.log(Level.SEVERE, String.format("Failed to parse file '%s'.",
+                                    upload.fileName()), result.cause());
+                                context.reroute(ERROR);
+                            }
+                        }));
                 });
             } else {
                 context.put(MESSAGE, NO_FILE_WAS_UPLOADED);
@@ -118,6 +146,7 @@ private void setRouterAPI(Router router) {
 
     /**
      * converts a throwables stack trace into a string.
+     *
      * @param throwable the throwable to be converted.
      * @return a textual representation of the throwables trace,
      * may be used in the app to display errors.
@@ -131,26 +160,28 @@ private String traceToText(Throwable throwable) {
     /**
      * Parses a file upload request, converting the excel payload into json and waits
      * for elasticsearch to complete indexing.
-     * @param buffer contains the excel file data
-     * @param params upload parameters
+     *
+     * @param buffer   contains the excel file data
+     * @param params   upload parameters
      * @param fileName the name of the uploaded file
-     * @param future callback on completed parse + indexing.
+     * @param future   callback on completed parse + indexing.
      */
     private void parse(Buffer buffer, MultiMap params, String fileName, Future<Integer> future) {
         vertx.<Integer>executeBlocking(blocking -> {
             try {
                 int columnRow = Integer.parseInt(params.get(OFFSET));
                 FileParser parser = new FileParser(buffer.getBytes(), columnRow, fileName);
-                vertx.eventBus().send(Configuration.INDEXING_ELASTICSEARCH,
-                        parser.toImportable(params.get(INDEX), getMappingByParams(params)),
-                        new DeliveryOptions().setSendTimeout(INDEXING_TIMEOUT),
-                        reply -> {
-                    if (reply.succeeded()) {
-                        blocking.complete(parser.getImportedItems());
-                    } else {
-                        blocking.fail(reply.cause());
-                    }
-                });
+                JsonObject data = parser.toImportable(params.get(INDEX), getMappingByParams(params))
+                    .put(UPLOAD_ID, params.get(UPLOAD_ID));
+
+                vertx.eventBus().send(INDEXING_ELASTICSEARCH, data, new DeliveryOptions().setSendTimeout(INDEXING_TIMEOUT),
+                    reply -> {
+                        if (reply.succeeded()) {
+                            blocking.complete(parser.getImportedItems());
+                        } else {
+                            blocking.fail(reply.cause());
+                        }
+                    });
             } catch (ParserException | NumberFormatException e) {
                 blocking.fail(new ParserException(e));
             }

src/main/java/com/codingchili/Model/ElasticWriter.java

@@ -8,6 +8,7 @@
 import java.util.logging.Logger;
 
 import static com.codingchili.Controller.Website.MAPPING;
+import static com.codingchili.Controller.Website.UPLOAD_ID;
 import static com.codingchili.Model.FileParser.ITEMS;
 
 /**
@@ -16,12 +17,14 @@
  *         Writes json data to elasticsearch for indexing.
  */
 public class ElasticWriter extends AbstractVerticle {
-    public static final int INDEXING_TIMEOUT = 300000;
+    public static final int INDEXING_TIMEOUT = 3000000;
     private static final String INDEX = "index";
     private static final String BULK = "/_bulk";
     private static final int POLL = 5000;
     private static final int MAX_BATCH = 255;
     public static final String ES_STATUS = "es-status";
+    private static final String PROGRESS = "progress";
+    public static final String IMPORT_PROGRESS = "import.progress";
     private static boolean connected = false;
     private static String version = "";
     private Logger logger = Logger.getLogger(getClass().getName());
@@ -90,6 +93,10 @@ private Future<Void> submitForIndexing(JsonArray items, JsonObject data, int cur
                             String.format("Submitted items [%d -> %d] of %d with result [%d] %s into '%s' [%.1f%%]",
                                     current, max - 1, items.size(), response.statusCode(), response.statusMessage(), index, percent));
 
+                    vertx.eventBus().publish(IMPORT_PROGRESS, new JsonObject()
+                        .put(PROGRESS, percent)
+                        .put(UPLOAD_ID, data.getString(UPLOAD_ID)));
+
                     future.complete();
                 })).exceptionHandler(exception -> future.fail(exception.getMessage()))
                 .end(bulkQuery(items, index, mapping, max, current));
@@ -110,18 +117,17 @@ private HttpClientRequest post(String path) {
      * @return a payload encoded as json-lines.
      */
     private String bulkQuery(JsonArray list, String index, String mapping, int max, int current) {
-        String query = "";
-        JsonObject header = new JsonObject()
+        StringBuilder query = new StringBuilder();
+        String header = new JsonObject()
                 .put("index", new JsonObject()
                         .put("_index", index)
-                        .put("_type", mapping));
+                        .put("_type", mapping)).encode() + "\n";
 
         for (int i = current; i < max; i++) {
-            query += header.encode() + "\n";
-            query += list.getJsonObject(i).encode() + "\n";
+            query.append(header).append(list.getJsonObject(i).encode()).append("\n");
         }
 
-        return query;
+        return query.toString();
     }
 
     /**