elasticsearch.http: implemented onMessageQueueEmpty function

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/ElasticSearchDestination.java

@@ -113,4 +113,9 @@ protected void deinit() {
 		client.deinit();
 		options.deinit();
 	}
+
+	@Override
+	public void onMessageQueueEmpty() {
+	    client.onMessageQueueEmpty();
+	}
 }

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/client/ESClient.java

@@ -33,4 +33,5 @@ public interface ESClient {
 	void deinit();
 	boolean send(ESIndex index);
 	String getClusterName();
+	void onMessageQueueEmpty();
 }

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/client/esnative/ESNativeClient.java

@@ -143,4 +143,9 @@ protected void loadConfigFile(String cfgFile, Builder settingsBuilder) {
 	public boolean send(ESIndex index) {
 		return messageProcessor.send(index);
 	}
+
+    @Override
+    public void onMessageQueueEmpty() {
+
+    }
 }

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/client/http/ESHttpClient.java

@@ -166,4 +166,9 @@ public String getClusterName() {
 		}
 		return clusterName;
 	}
+
+    @Override
+    public void onMessageQueueEmpty() {
+        messageProcessor.onMessageQueueEmpty();
+    }
 }

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/messageprocessor/http/HttpBulkMessageProcessor.java

@@ -72,4 +72,9 @@ public boolean send(Index index) {
 		messageCounter++;
 		return true;
 	}
+
+    @Override
+    public void onMessageQueueEmpty() {
+        flush();
+    }
 }

modules/java-modules/elastic-v2/src/main/java/org/syslog_ng/elasticsearch_v2/messageprocessor/http/HttpMessageProcessor.java

@@ -60,4 +60,8 @@ public boolean send(ESIndex index) {
 		Index req = new Index.Builder(index.getFormattedMessage()).index(index.getIndex()).type(index.getType()).id(index.getId()).build();
 		return send(req);
 	}
+
+    public void onMessageQueueEmpty() {
+
+    }
 }