Adapt to changed decorators interface (#43)

* Providing a message decorator that uses pipelines.
* Making decorator configurable.
* Allow adding new messages by pipeline decorator.
* Adding changes related due to introduced listener.
* Adapt to naming changes, using easier forEach idiom.
* Changing decorator to work on SearchResponse instead of message list.

Author: dennisoelkers

src/main/java/org/graylog/plugins/pipelineprocessor/PipelineProcessorMessageDecorator.java

@@ -19,34 +19,33 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Multimap;
 import com.google.inject.assistedinject.Assisted;
 import org.graylog.plugins.pipelineprocessor.db.PipelineDao;
 import org.graylog.plugins.pipelineprocessor.db.PipelineService;
 import org.graylog.plugins.pipelineprocessor.processors.PipelineInterpreter;
 import org.graylog.plugins.pipelineprocessor.processors.listeners.NoopInterpreterListener;
 import org.graylog2.decorators.Decorator;
-import org.graylog2.indexer.results.ResultMessage;
 import org.graylog2.plugin.Message;
 import org.graylog2.plugin.configuration.ConfigurationRequest;
 import org.graylog2.plugin.configuration.fields.ConfigurationField;
 import org.graylog2.plugin.configuration.fields.DropdownField;
-import org.graylog2.plugin.decorators.MessageDecorator;
+import org.graylog2.plugin.decorators.SearchResponseDecorator;
+import org.graylog2.rest.models.messages.responses.ResultMessageSummary;
+import org.graylog2.rest.resources.search.responses.SearchResponse;
 
 import javax.inject.Inject;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-public class PipelineProcessorMessageDecorator implements MessageDecorator {
+public class PipelineProcessorMessageDecorator implements SearchResponseDecorator {
     private static final String CONFIG_FIELD_PIPELINE = "pipeline";
 
     private final PipelineInterpreter pipelineInterpreter;
     private final ImmutableSet<String> pipelines;
 
-    public interface Factory extends MessageDecorator.Factory {
+    public interface Factory extends SearchResponseDecorator.Factory {
         @Override
         PipelineProcessorMessageDecorator create(Decorator decorator);
 
@@ -57,7 +56,7 @@ public interface Factory extends MessageDecorator.Factory {
         Descriptor getDescriptor();
     }
 
-    public static class Config implements MessageDecorator.Config {
+    public static class Config implements SearchResponseDecorator.Config {
         private final PipelineService pipelineService;
 
         @Inject
@@ -81,7 +80,7 @@ public ConfigurationRequest getRequestedConfiguration() {
         };
     }
 
-    public static class Descriptor extends MessageDecorator.Descriptor {
+    public static class Descriptor extends SearchResponseDecorator.Descriptor {
         public Descriptor() {
             super("Pipeline Processor Decorator", false, "http://docs.graylog.org/en/2.0/pages/pipelines.html", "Pipeline Processor Decorator");
         }
@@ -100,26 +99,25 @@ public PipelineProcessorMessageDecorator(PipelineInterpreter pipelineInterpreter
     }
 
     @Override
-    public List<ResultMessage> apply(List<ResultMessage> resultMessages) {
-        final List<ResultMessage> results = new ArrayList<>();
+    public SearchResponse apply(SearchResponse searchResponse) {
+        final List<ResultMessageSummary> results = new ArrayList<>();
         if (pipelines.isEmpty()) {
-            return resultMessages;
+            return searchResponse;
         }
-        resultMessages.forEach((inMessage) -> {
-            final Message message = inMessage.getMessage();
+        searchResponse.messages().forEach((inMessage) -> {
+            final Message message = new Message(inMessage.message());
             final List<Message> additionalCreatedMessages = pipelineInterpreter.processForPipelines(message,
                     message.getId(),
                     pipelines,
                     new NoopInterpreterListener());
-            final ResultMessage outMessage = ResultMessage.createFromMessage(message, inMessage.getIndex(), inMessage.getHighlightRanges());
 
-            results.add(outMessage);
+            results.add(ResultMessageSummary.create(inMessage.highlightRanges(), message.getFields(), inMessage.index()));
             additionalCreatedMessages.forEach((additionalMessage) -> {
                 // TODO: pass proper highlight ranges. Need to rebuild them for new messages.
-                results.add(ResultMessage.createFromMessage(additionalMessage, "[created from decorator]", ImmutableMultimap.of()));
+                results.add(ResultMessageSummary.create(ImmutableMultimap.of(), additionalMessage.getFields(), "[created from decorator]"));
             });
         });
 
-        return results;
+        return searchResponse.toBuilder().messages(results).build();
     }
 }

src/main/java/org/graylog/plugins/pipelineprocessor/PipelineProcessorModule.java

@@ -47,6 +47,8 @@ protected void configure() {
 
         install(new ProcessorFunctionsModule());
 
-        installMessageDecorator(messageDecoratorBinder(), PipelineProcessorMessageDecorator.class, PipelineProcessorMessageDecorator.Factory.class);
+        installSearchResponseDecorator(searchResponseDecoratorBinder(),
+                PipelineProcessorMessageDecorator.class,
+                PipelineProcessorMessageDecorator.Factory.class);
     }
 }