Skip to content

Commit fe0a5a8

Browse files
author
chilimannen
committed
Add back-pressure to imports + improve logging + always open browser
1 parent 511bbe2 commit fe0a5a8

File tree

4 files changed

+72
-43
lines changed

4 files changed

+72
-43
lines changed

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import com.codingchili.Model.ParserException;
1515

1616
import io.vertx.core.*;
17+
import io.vertx.core.eventbus.MessageConsumer;
1718

1819
import static com.codingchili.Model.ElasticWriter.ES_STATUS;
1920

@@ -46,13 +47,15 @@ public ApplicationLauncher(String[] args) {
4647
if (args.length == 2) {
4748
importFile(getFileName(), getIndexName());
4849
} else {
49-
vertx.eventBus().consumer(ES_STATUS, message -> {
50-
logger.info("Attempting to open browser..");
50+
MessageConsumer<?> consumer = vertx.eventBus().consumer(ES_STATUS);
51+
consumer.handler(message -> {
52+
logger.info(String.format("Attempting to open browser.. [ES connected=%s]", message.body().toString()));
5153
try {
5254
Desktop.getDesktop().browse(new URI(Configuration.getWebsiteURL()));
5355
} catch (IOException | URISyntaxException e) {
5456
logger.warning(e.getMessage());
5557
}
58+
consumer.pause();
5659
});
5760
}
5861
} else {

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -83,20 +83,24 @@ private void setRouterAPI(Router router) {
8383
Iterator<FileUpload> iterator = context.fileUploads().iterator();
8484

8585
if (iterator.hasNext()) {
86+
logger.info("Receiving uploaded file.. ");
8687
FileUpload upload = context.fileUploads().iterator().next();
8788

8889
vertx.fileSystem().readFile(upload.uploadedFileName(), file -> {
8990
parse(file.result(), context.request().params(), upload.fileName(),
9091
Future.<Integer>future().setHandler(result -> {
9192
if (result.succeeded()) {
92-
logger.info(String.format("Imported file %s successfully.", upload.fileName()));
93-
context.put(INDEX, context.request().params().get(INDEX));
93+
String index = context.request().params().get(INDEX);
94+
logger.info(String.format("Imported file '%s' successfully into '%s'.",
95+
upload.fileName(), index));
96+
97+
context.put(INDEX, index);
9498
context.put(FILE, upload.fileName());
9599
context.put(IMPORTED, result.result());
96100
context.reroute(DONE);
97101
} else {
98102
context.put(MESSAGE, traceToText(result.cause()));
99-
logger.log(Level.SEVERE, String.format("Failed to parse file %s.",
103+
logger.log(Level.SEVERE, String.format("Failed to parse file '%s'.",
100104
upload.fileName()), result.cause());
101105
context.reroute(ERROR);
102106
}

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

Lines changed: 59 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,11 @@
11
package com.codingchili.Model;
22

3-
import java.util.concurrent.CountDownLatch;
4-
import java.util.logging.Logger;
5-
6-
import io.vertx.core.AbstractVerticle;
7-
import io.vertx.core.Context;
8-
import io.vertx.core.Future;
9-
import io.vertx.core.Vertx;
3+
import io.vertx.core.*;
104
import io.vertx.core.json.JsonArray;
115
import io.vertx.core.json.JsonObject;
126

7+
import java.util.logging.Logger;
8+
139
import static com.codingchili.Model.FileParser.ITEMS;
1410

1511
/**
@@ -49,35 +45,57 @@ private void startSubmitListener() {
4945
vertx.eventBus().consumer(Configuration.INDEXING_ELASTICSEARCH, handler -> {
5046
JsonObject data = (JsonObject) handler.body();
5147
JsonArray items = data.getJsonArray(ITEMS);
52-
String index = data.getString(INDEX);
53-
CountDownLatch latch = new CountDownLatch(getBatchCount(items.size()));
48+
Future<Void> starter = Future.future();
49+
Future<Void> next = starter;
5450

55-
// performs one bulk request for each bucket of MAX_BATCH
51+
// performs one bulk request for each bucket of MAX_BATCH serially.
5652
for (int i = 0; i < items.size(); i += MAX_BATCH) {
57-
final int max = ((i + MAX_BATCH < items.size()) ? i + MAX_BATCH : items.size());
5853
final int current = i;
59-
vertx.createHttpClient().post(
60-
Configuration.getElasticPort(), Configuration.getElasticHost(), index + BULK)
61-
.handler(response -> {
62-
logger.info(String.format("Submitted items [%d -> %d] with result [%d] %s",
63-
current, max - 1, response.statusCode(), response.statusMessage()));
64-
65-
response.bodyHandler(body -> {
66-
latch.countDown();
67-
68-
// complete successfully when all buckets are inserted.
69-
if (latch.getCount() == 0) {
70-
handler.reply(null);
71-
}
72-
});
73-
}).exceptionHandler(exception -> handler.fail(500, exception.getMessage()))
74-
.end(bulkQuery(items, index, max, current));
54+
final int max = ((current + MAX_BATCH < items.size()) ? current + MAX_BATCH : items.size());
55+
next = next.compose(v -> submitForIndexing(items, data.getString(INDEX), current, max));
7556
}
57+
58+
// when the final submission completes complete the handler.
59+
next.setHandler((v) -> {
60+
if (v.succeeded()) {
61+
handler.reply(null);
62+
} else {
63+
handler.fail(500, v.cause().getMessage());
64+
}
65+
});
66+
starter.complete();
7667
});
7768
}
7869

70+
/**
71+
* Submits a subset of the given json array for indexing.
72+
*
73+
* @param items items to be indexed
74+
* @param index the name of the index to use
75+
* @param current the low bound for the given json items to import
76+
* @param max the high bound for the given json items to import
77+
* @return a future completed when the indexing of the specified elements have completed.
78+
*/
79+
private Future<Void> submitForIndexing(JsonArray items, String index, int current, int max) {
80+
Future<Void> future = Future.future();
81+
vertx.createHttpClient().post(
82+
Configuration.getElasticPort(), Configuration.getElasticHost(), index + BULK)
83+
.handler(response -> response.bodyHandler(body -> {
84+
85+
float percent = (max * 1.0f / items.size()) * 100;
86+
logger.info(
87+
String.format("Submitted items [%d -> %d] of %d with result [%d] %s into '%s' [%.1f%%]",
88+
current, max - 1, items.size(), response.statusCode(), response.statusMessage(), index, percent));
89+
90+
future.complete();
91+
})).exceptionHandler(exception -> future.fail(exception.getMessage()))
92+
.end(bulkQuery(items, index, max, current));
93+
return future;
94+
}
95+
7996
/**
8097
* Determines the number of batches to bulk insert.
98+
*
8199
* @param size the total number of items to insert
82100
* @return the number of batches required to submit the size
83101
*/
@@ -87,9 +105,10 @@ private int getBatchCount(int size) {
87105

88106
/**
89107
* Builds a bulk query for insertion into elasticsearch
90-
* @param list full list of all elements
91-
* @param index the current item anchor
92-
* @param max max upper bound of items to include in the bulk
108+
*
109+
* @param list full list of all elements
110+
* @param index the current item anchor
111+
* @param max max upper bound of items to include in the bulk
93112
* @param current lower bound of items to include in the bulk
94113
* @return a payload encoded as json-lines.
95114
*/
@@ -111,21 +130,23 @@ private String bulkQuery(JsonArray list, String index, int max, int current) {
111130
/**
112131
* Polls the elasticsearch server for version information. Sets connected if the server
113132
* is available.
133+
*
114134
* @param id the id of the timer that triggered the request, not used.
115135
*/
116136
private void pollElasticServer(Long id) {
117137
vertx.createHttpClient().get(Configuration.getElasticPort(), Configuration.getElasticHost(), "/",
118138
response -> response.bodyHandler(buffer -> {
119-
version = buffer.toJsonObject().getJsonObject("version").getString("number");
120-
if (!connected) {
121-
logger.info(String.format("Connected to elasticsearch server %s at %s:%d",
122-
version, Configuration.getElasticHost(), Configuration.getElasticPort()));
123-
connected = true;
124-
vertx.eventBus().send(ES_STATUS, connected);
125-
}
126-
})).exceptionHandler(event -> {
139+
version = buffer.toJsonObject().getJsonObject("version").getString("number");
140+
if (!connected) {
141+
logger.info(String.format("Connected to elasticsearch server %s at %s:%d",
142+
version, Configuration.getElasticHost(), Configuration.getElasticPort()));
143+
connected = true;
144+
vertx.eventBus().send(ES_STATUS, connected);
145+
}
146+
})).exceptionHandler(event -> {
127147
connected = false;
128148
logger.severe(event.getMessage());
149+
vertx.eventBus().send(ES_STATUS, connected);
129150
}).end();
130151
}
131152

‎src/main/java/com/codingchili/Model/FileParser.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ public class FileParser {
3434
* @param offset row number containing column titles.
3535
*/
3636
public FileParser(byte[] bytes, int offset, String fileName) throws ParserException {
37+
logger.info(String.format("Parsing file '%s' using titles from excel row %d..", fileName, offset));
3738
offset -= 1; // convert excel row name to index.
3839
try {
3940
Workbook workbook = getWorkbook(bytes, fileName);

0 commit comments

Comments
 (0)