Unregister PipelineInterpreter from event bus

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

@@ -17,7 +17,6 @@
 package org.graylog.plugins.pipelineprocessor;
 
 import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultimap;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.assistedinject.Assisted;
@@ -123,6 +122,8 @@ public SearchResponse apply(SearchResponse searchResponse) {
             });
         });
 
+        pipelineInterpreter.stop();
+
         return searchResponse.toBuilder().messages(results).build();
     }
 }

src/main/java/org/graylog/plugins/pipelineprocessor/processors/PipelineInterpreter.java

@@ -85,6 +85,7 @@ public class PipelineInterpreter implements MessageProcessor {
     private final MetricRegistry metricRegistry;
     private final ScheduledExecutorService scheduler;
     private final Meter filteredOutMessages;
+    private EventBus serverEventBus;
 
     private final AtomicReference<ImmutableMap<String, Pipeline>> currentPipelines = new AtomicReference<>(ImmutableMap.of());
     private final AtomicReference<ImmutableSetMultimap<String, Pipeline>> streamPipelineConnections = new AtomicReference<>(ImmutableSetMultimap.of());
@@ -107,13 +108,22 @@ public PipelineInterpreter(RuleService ruleService,
         this.metricRegistry = metricRegistry;
         this.scheduler = scheduler;
         this.filteredOutMessages = metricRegistry.meter(name(ProcessBufferProcessor.class, "filteredOutMessages"));
+        this.serverEventBus = serverEventBus;
 
         // listens to cluster wide Rule, Pipeline and pipeline stream connection changes
         serverEventBus.register(this);
 
         reload();
     }
 
+    /*
+     * Allow to unregister PipelineInterpreter from the event bus, allowing the object to be garbage collected.
+     * This is needed in some classes, when new PipelineInterpreter instances are created per request.
+     */
+    public void stop() {
+        serverEventBus.unregister(this);
+    }
+
     // this should not run in parallel
     private synchronized void reload() {
         // read all rules and compile them

src/main/java/org/graylog/plugins/pipelineprocessor/rest/SimulatorResource.java

@@ -82,6 +82,8 @@ public SimulationResponse simulate(@ApiParam(name = "simulation", required = tru
         for (Message processedMessage : processedMessages) {
             simulationResults.add(ResultMessageSummary.create(null, processedMessage.getFields(), ""));
         }
+
+        pipelineInterpreter.stop();
         return SimulationResponse.create(simulationResults,
                                          pipelineInterpreterTracer.getExecutionTrace(),
                                          pipelineInterpreterTracer.took());