:register_method:	true
:run_method:		true
:plugintype:    	input
:pluginclass:   	Inputs
:pluginname:    	example
:pluginnamecap: 	Example
:plugintypecap:   Input
:sversion:        '0.0.1'

:pluginrepo:      https://github.com/logstash-plugins/logstash-input-java_input_example[example input plugin]


[[java-input-plugin]]
=== How to write a Java input plugin

// Pulls in shared section: Setting Up Environment
include::include/javapluginsetup.asciidoc[]

[float]
=== Code the plugin

The example input plugin generates a configurable number of simple events before
terminating. Let's look at the main class in the  example input.
 
[source,java]
----- 
@LogstashPlugin(name="java_input_example")
public class JavaInputExample implements Input {

    public static final PluginConfigSpec<Long> EVENT_COUNT_CONFIG =
            PluginConfigSpec.numSetting("count", 3);

    public static final PluginConfigSpec<String> PREFIX_CONFIG =
            PluginConfigSpec.stringSetting("prefix", "message");

    private String id;
    private long count;
    private String prefix;
    private final CountDownLatch done = new CountDownLatch(1);
    private volatile boolean stopped;


    public JavaInputExample(String id, Configuration config, Context context) {
            this.id = id;
        count = config.get(EVENT_COUNT_CONFIG);
        prefix = config.get(PREFIX_CONFIG);
    }

    @Override
    public void start(Consumer<Map<String, Object>> consumer) {
        int eventCount = 0;
        try {
            while (!stopped && eventCount < count) {
                eventCount++;
                consumer.accept.push(Collections.singletonMap("message",
                        prefix + " " + StringUtils.center(eventCount + " of " + count, 20)));
            }
        } finally {
            stopped = true;
            done.countDown();
        }
    }

    @Override
    public void stop() {
        stopped = true; // set flag to request cooperative stop of input
    }

    @Override
    public void awaitStop() throws InterruptedException {
        done.await(); // blocks until input has stopped
    }

    @Override
    public Collection<PluginConfigSpec<?>> configSchema() {
        return Arrays.asList(EVENT_COUNT_CONFIG, PREFIX_CONFIG);
    }

    @Override
    public String getId() {
        return this.id;
    }
}
-----

Let's step through and examine each part of that class.

[float]
==== Class declaration

[source,java]
-----
@LogstashPlugin(name="java_input_example")
public class JavaInputExample implements Input {
-----

Notes about the class declaration:

* All Java plugins must be annotated with the `@LogstashPlugin` annotation. Additionally:
** The `name` property of the annotation must be supplied and defines the name of the plugin as it will be used
   in the Logstash pipeline definition. For example, this input would be referenced in the input section of the
   Logstash pipeline defintion as `input { java_input_example => { .... } }`
** The value of the `name` property must match the name of the class excluding casing and underscores.
* The class must implement the `co.elastic.logstash.api.Input` interface.
* Java plugins may not be created in the `org.logstash` or `co.elastic.logstash` packages to prevent potential
clashes with classes in Logstash itself.

[float]
==== Plugin settings

The snippet below contains both the setting definition and the method referencing it.

[source,java]
-----
public static final PluginConfigSpec<Long> EVENT_COUNT_CONFIG =
        PluginConfigSpec.numSetting("count", 3);

public static final PluginConfigSpec<String> PREFIX_CONFIG =
        PluginConfigSpec.stringSetting("prefix", "message");

@Override
public Collection<PluginConfigSpec<?>> configSchema() {
    return Arrays.asList(EVENT_COUNT_CONFIG, PREFIX_CONFIG);
}
-----

The `PluginConfigSpec` class allows developers to specify the settings that a
plugin supports complete with setting  name, data type, deprecation status,
required status, and default value. In this example, the `count` setting defines
the number of events that will be generated and the `prefix` setting defines an
optional prefix to include in the event field. Neither setting is required and
if it is not explicitly set, the settings default to `3` and  `message`,
respectively.

The `configSchema` method must return a list of all settings that the plugin
supports. In a future phase of the Java plugin project, the Logstash execution
engine will validate that all required settings are present and that no
unsupported settings are present.

[float]
==== Constructor and initialization

[source,java]
-----
private String id;
private long count;
private String prefix;

public JavaInputExample(String id, Configuration config, Context context) {
    this.id = id;
    count = config.get(EVENT_COUNT_CONFIG);
    prefix = config.get(PREFIX_CONFIG);
}
-----

All Java input plugins must have a constructor taking a `String` id and
`Configuration` and `Context` argument. This is the constructor that will be
used to instantiate them at runtime. The retrieval and validation of all plugin
settings should occur in this constructor. In this example, the values of the
two plugin settings are retrieved and stored in local variables for later use in
the `start` method. 

Any additional initialization may occur in the constructor as well. If there are
any unrecoverable errors encountered in the configuration or initialization of
the input plugin, a descriptive exception should be thrown. The exception will
be logged and will prevent Logstash from starting.

[float]
==== Start method

[source,java]
-----
@Override
public void start(Consumer<Map<String, Object>> consumer) {
    int eventCount = 0;
    try {
        while (!stopped && eventCount < count) {
            eventCount++;
            consumer.accept.push(Collections.singletonMap("message",
                    prefix + " " + StringUtils.center(eventCount + " of " + count, 20)));
        }
    } finally {
        stopped = true;
        done.countDown();
    }
}
-----

The `start` method begins the event-producing loop in an input. Inputs are flexible and may produce events through
many different mechanisms including:

 * a pull mechanism such as periodic queries of external database
 * a push mechanism such as events sent from clients to a local network port
 * a timed computation such as a heartbeat
 * any other mechanism that produces a useful stream of events. Event streams may be either finite or infinite. 
If the input produces an infinite stream of events, this method should loop until a stop request is made through
the `stop` method. If the input produces a finite stream of events, this method should terminate when the last 
event in the stream is produced or a stop request is made, whichever comes first.

Events should be constructed as instances of `Map<String, Object>` and pushed into the event pipeline via the
`Consumer<Map<String, Object>>.accept()` method. To reduce allocations and GC pressure, inputs may reuse the same
map instance by modifying its fields between calls to `Consumer<Map<String, Object>>.accept()` because the event
pipeline will create events based on a copy of the map's data.

[float]
==== Stop and awaitStop methods

[source,java]
-----
private final CountDownLatch done = new CountDownLatch(1);
private volatile boolean stopped;

@Override
public void stop() {
    stopped = true; // set flag to request cooperative stop of input
}

@Override
public void awaitStop() throws InterruptedException {
    done.await(); // blocks until input has stopped
}
-----

The `stop` method notifies the input to stop producing events. The stop
mechanism may be implemented in any way that honors the API contract though a
`volatile boolean` flag works well for many use cases.

Inputs stop both asynchronously and cooperatively. Use the `awaitStop` method to
block until the input has  completed the stop process. Note that this method
should **not** signal the input to stop as the `stop` method does. The
awaitStop mechanism may be implemented in any way that honors the API contract
though a `CountDownLatch` works well for many use cases.

[float]
==== getId method

[source,java]
-----
@Override
public String getId() {
    return id;
}
-----

For input plugins, the `getId` method should always return the id that was provided to the plugin through its
constructor at instantiation time.

[float]
==== Unit tests
Lastly, but certainly not least importantly, unit tests are strongly encouraged.
The example input plugin includes an
https://github.com/logstash-plugins/logstash-input-java_input_example/blob/main/src/test/java/org/logstashplugins/JavaInputExampleTest.java[example unit
test] that you can use as a template for your own.

// Pulls in shared section about Packaging and Deploying
include::include/javapluginpkg.asciidoc[]

[float]
=== Running Logstash with the Java input plugin

The following is a minimal Logstash configuration that can be used to test that
the Java input plugin is correctly installed and functioning.

[source,java]
-----
input {
  java_input_example {}
}
output {
  stdout { codec => rubydebug }
}
-----

Copy the above Logstash configuration to a file such as `java_input.conf`. 
Start {ls} with:

[source,shell]
-----
bin/logstash -f /path/to/java_input.conf
-----


The expected Logstash output (excluding initialization) with the configuration above is:

[source,txt]
-----
{
      "@version" => "1",
       "message" => "message        1 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}
{
      "@version" => "1",
       "message" => "message        2 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}
{
      "@version" => "1",
       "message" => "message        3 of 3       ",
    "@timestamp" => yyyy-MM-ddThh:mm:ss.SSSZ
}
-----

[float]
=== Feedback

If you have any feedback on Java plugin support in Logstash, please comment on our 
https://github.com/elastic/logstash/issues/9215[main Github issue] or post in the 
https://discuss.elastic.co/c/logstash[Logstash forum].

:pluginrepo!:
:sversion!:
:plugintypecap!:
:pluginnamecap!:
:pluginname!:
:pluginclass!:
:plugintype!: