@Override
    public Messages process(Messages messages) {
        try (Timer.Context ignored = executionTime.time()) {
            final State latestState = stateUpdater.getLatestState();
            if (latestState.enableRuleMetrics()) {
                return process(messages, new RuleMetricsListener(metricRegistry), latestState);
            }
            return process(messages, new NoopInterpreterListener(), latestState);
        }
    }
@Test
    public void testDroppedMessageWillHaltProcessingAfterCurrentStage() {
        final RuleService ruleService = mock(MongoDbRuleService.class);
        when(ruleService.loadAll()).thenReturn(ImmutableList.of(
                RULE_SET_FIELD.apply("1-a"),
                RULE_SET_FIELD.apply("1-b"),
                RULE_SET_FIELD.apply("2-a"),
                RULE_SET_FIELD.apply("2-b"),
                RULE_DROP_MESSAGE
        ));

        final PipelineService pipelineService = mock(MongoDbPipelineService.class);
        when(pipelineService.loadAll()).thenReturn(ImmutableList.of(
                PipelineDao.create("p1", "title1", "description",
                        "pipeline \"pipeline1\"\n" +
                                "stage 0 match pass\n" +
                                "    rule \"1-a\";\n" +
                                "    rule \"drop_message\";\n" +
                                "stage 1 match pass\n" +
                                "    rule \"1-b\";\n" +
                                "end\n",
                        Tools.nowUTC(),
                        null),
                PipelineDao.create("p2", "title2", "description",
                        "pipeline \"pipeline2\"\n" +
                                "stage 0 match pass\n" +
                                "    rule \"2-a\";\n" +
                                "stage 1 match pass\n" +
                                "    rule \"2-b\";\n" +
                                "end\n",
                        Tools.nowUTC(),
                        null)
        ));

        final Map<String, Function<?>> functions = ImmutableMap.of(
                SetField.NAME, new SetField(),
                DropMessage.NAME, new DropMessage()
        );
        final PipelineInterpreter interpreter = createPipelineInterpreter(ruleService, pipelineService, functions);

        final Messages processed = interpreter.process(messageInDefaultStream("message", "test"));

        assertThat(processed).isInstanceOf(MessageCollection.class);

        // Use MessageCollection#source here to get access to the unfiltered messages
        final List<Message> messages = ImmutableList.copyOf(((MessageCollection) processed).source());
        assertThat(messages).hasSize(1);

        final Message actualMessage = messages.get(0);

        assertThat(actualMessage.getFilterOut()).isTrue();

        // Even though "drop_message" has been called in one of the stages, all stages of the same number should
        // have been executed
        assertThat(actualMessage.getFieldAs(String.class, "1-a")).isEqualTo("value");
        assertThat(actualMessage.getFieldAs(String.class, "2-a")).isEqualTo("value");

        // The second stage in both pipelines should not have been executed due to the "drop_message" call
        assertThat(actualMessage.getField("1-b")).isNull();
        assertThat(actualMessage.getField("2-b")).isNull();
    }