Update input plugin for beta version of the Java API

Update input plugin for beta version of the Java API

Author: danhermann

README.md

@@ -6,4 +6,4 @@ This is a Java plugin for [Logstash](https://github.com/elastic/logstash).
 
 It is fully free and fully open source. The license is Apache 2.0, meaning you are free to use it however you want.
 
-The documentation for Logstash Java plugins is available [here](https://www.elastic.co/guide/en/logstash/6.6/contributing-java-plugin.html).
+The documentation for Logstash Java plugins is available [here](https://www.elastic.co/guide/en/logstash/6.7/contributing-java-plugin.html).

build.gradle

@@ -8,9 +8,9 @@ apply plugin: 'maven'
 apply plugin: "distribution"
 
 group 'org.logstash.javaapi'
-version "${new File("VERSION").text.trim()}"
+version "${file("VERSION").text.trim()}"
 
-description = "Example Java filter implementation"
+description = "Example Java input implementation"
 
 sourceCompatibility = 1.8
 targetCompatibility = 1.8

src/main/java/org/logstash/javaapi/JavaInputExample.java

@@ -2,41 +2,44 @@
 
 import co.elastic.logstash.api.Configuration;
 import co.elastic.logstash.api.Context;
+import co.elastic.logstash.api.Input;
 import co.elastic.logstash.api.LogstashPlugin;
 import co.elastic.logstash.api.PluginConfigSpec;
-import co.elastic.logstash.api.v0.Input;
 import org.apache.commons.lang3.StringUtils;
-import org.logstash.execution.queue.QueueWriter;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.function.Consumer;
 
 // class name must match plugin name
 @LogstashPlugin(name="java_input_example")
 public class JavaInputExample implements Input {
 
     public static final PluginConfigSpec<Long> EVENT_COUNT_CONFIG =
-            Configuration.numSetting("count", 3);
+            PluginConfigSpec.numSetting("count", 3);
 
     public static final PluginConfigSpec<String> PREFIX_CONFIG =
-            Configuration.stringSetting("prefix", "message");
+            PluginConfigSpec.stringSetting("prefix", "message");
 
+    private String id;
     private long count;
     private String prefix;
     private final CountDownLatch done = new CountDownLatch(1);
     private volatile boolean stopped;
 
-    // all plugins must provide a constructor that accepts Configuration and Context
-    public JavaInputExample(Configuration config, Context context) {
+    // all plugins must provide a constructor that accepts id, Configuration, and Context
+    public JavaInputExample(String id, Configuration config, Context context) {
         // constructors should validate configuration options
+        this.id = id;
         count = config.get(EVENT_COUNT_CONFIG);
         prefix = config.get(PREFIX_CONFIG);
     }
 
     @Override
-    public void start(QueueWriter queueWriter) {
+    public void start(Consumer<Map<String, Object>> consumer) {
 
         // The start method should push Map<String, Object> instances to the supplied QueueWriter
         // instance. Those will be converted to Event instances later in the Logstash event
@@ -51,7 +54,7 @@ public void start(QueueWriter queueWriter) {
         try {
             while (!stopped && eventCount < count) {
                 eventCount++;
-                queueWriter.push(Collections.singletonMap("message",
+                consumer.accept(Collections.singletonMap("message",
                         prefix + " " + StringUtils.center(eventCount + " of " + count, 20)));
             }
         } finally {
@@ -75,4 +78,9 @@ public Collection<PluginConfigSpec<?>> configSchema() {
         // should return a list of all configuration options for this plugin
         return Arrays.asList(EVENT_COUNT_CONFIG, PREFIX_CONFIG);
     }
+
+    @Override
+    public String getId() {
+        return this.id;
+    }
 }

src/test/java/org/logstash/javaapi/JavaInputExampleTest.java

@@ -4,12 +4,13 @@
 import org.apache.commons.lang3.StringUtils;
 import org.junit.Assert;
 import org.junit.Test;
-import org.logstash.execution.queue.QueueWriter;
+import org.logstash.plugins.ConfigurationImpl;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 
 public class JavaInputExampleTest {
 
@@ -20,32 +21,33 @@ public void testJavaInputExample() {
         Map<String, Object> configValues = new HashMap<>();
         configValues.put(JavaInputExample.PREFIX_CONFIG.name(), prefix);
         configValues.put(JavaInputExample.EVENT_COUNT_CONFIG.name(), eventCount);
-        Configuration config = new Configuration(configValues);
-        JavaInputExample input = new JavaInputExample(config, null);
-        TestQueueWriter testQueueWriter = new TestQueueWriter();
-        input.start(testQueueWriter);
+        Configuration config = new ConfigurationImpl(configValues);
+        JavaInputExample input = new JavaInputExample("test-id", config, null);
+        TestConsumer testConsumer = new TestConsumer();
+        input.start(testConsumer);
 
-        List<Map<String, Object>> events = testQueueWriter.getEvents();
+        List<Map<String, Object>> events = testConsumer.getEvents();
         Assert.assertEquals(eventCount, events.size());
         for (int k = 1; k <= events.size(); k++) {
             Assert.assertEquals(prefix + " " + StringUtils.center(k + " of " + eventCount, 20),
                     events.get(k - 1).get("message"));
         }
     }
-}
 
-class TestQueueWriter implements QueueWriter {
+    private static class TestConsumer implements Consumer<Map<String, Object>> {
 
-    private List<Map<String, Object>> events = new ArrayList<>();
+        private List<Map<String, Object>> events = new ArrayList<>();
 
-    @Override
-    public void push(Map<String, Object> event) {
-        synchronized (this) {
-            events.add(event);
+        @Override
+        public void accept(Map<String, Object> event) {
+            synchronized (this) {
+                events.add(event);
+            }
         }
-    }
 
-    public List<Map<String, Object>> getEvents() {
-        return events;
+        public List<Map<String, Object>> getEvents() {
+            return events;
+        }
     }
+
 }