public void init(AtomicReference<PipelineRuleOutputFilterState> activeState) {
        reload(activeState, ReloadTrigger.empty());
    }
@Test
    void init(MessageFactory messageFactory) {
        final var defaultStreamDestinationFilter = StreamDestinationFilterRuleDTO.builder()
                .id("54e3deadbeefdeadbeef0001")
                .title("Test 1")
                .streamId(defaultStream.getId())
                .destinationType("indexer")
                .rule(RuleBuilder.builder().build()) // Not needed because we mock the RuleBuilderService
                .build();
        final var testStreamDestinationFilter = StreamDestinationFilterRuleDTO.builder()
                .id("54e3deadbeefdeadbeef0002")
                .title("Test 2")
                .streamId(testStream.getId())
                .destinationType("indexer")
                .rule(RuleBuilder.builder().build()) // Not needed because we mock the RuleBuilderService
                .build();
        final var defaultStreamRuleDao = RuleDao.builder()
                .id(defaultStreamDestinationFilter.id())
                .title("does-not-matter")
                .source("""
                        rule "%s"
                        when has_field(field : "source")
                        then
                          __remove_from_stream_destination__(stream_id : "%s", destination_type : "indexer");
                        end
                        """.formatted(ruleTitle(defaultStreamDestinationFilter), defaultStream.getId()))
                .build();
        final var testStreamRuleDao = RuleDao.builder()
                .id(testStreamDestinationFilter.id())
                .title("does-not-matter")
                .source("""
                        rule "%s"
                        when has_field(field : "source")
                        then
                          __remove_from_stream_destination__(stream_id : "%s", destination_type : "indexer");
                        end
                        """.formatted(ruleTitle(testStreamDestinationFilter), testStream.getId()))
                .build();

        doAnswer((Answer<Void>) invocation -> {
            final Consumer<StreamDestinationFilterService.GroupByStreamResult> consumer = invocation.getArgument(0);
            // The consumer must be called once for each stream group
            consumer.accept(new StreamDestinationFilterService.GroupByStreamResult(defaultStream.getId(), Set.of(defaultStreamDestinationFilter)));
            consumer.accept(new StreamDestinationFilterService.GroupByStreamResult(testStream.getId(), Set.of(testStreamDestinationFilter)));
            return null;
        }).when(filterService).forEachEnabledFilterGroupedByStream(any());

        when(resolverFactory.create(any(), any())).thenReturn(createResolver(List.of(defaultStreamRuleDao, testStreamRuleDao)));

        // Mock each rule source generator call to return the correct source for the given rule
        when(ruleBuilderService.generateRuleSource(eq(ruleTitle(defaultStreamDestinationFilter)), any(RuleBuilder.class), anyBoolean()))
                .thenReturn(defaultStreamRuleDao.source());
        when(ruleBuilderService.generateRuleSource(eq(ruleTitle(testStreamDestinationFilter)), any(RuleBuilder.class), anyBoolean()))
                .thenReturn(testStreamRuleDao.source());

        final var stateUpdater = new PipelineRuleOutputFilterStateUpdater(
                filterService,
                (pipelines, destinations, activeStreams) -> new PipelineRuleOutputFilterState(
                        pipelines,
                        destinations,
                        activeStreams,
                        new MetricRegistry(),
                        1,
                        true
                ),
                Map.of("indexer", mock(FilteredMessageOutput.class)),
                ruleBuilderService,
                resolverFactory,
                ruleParser,
                new MetricRegistry()
        );

        final var activeState = new AtomicReference<PipelineRuleOutputFilterState>();
        stateUpdater.init(activeState);

        assertThat(activeState.get()).isNotNull().satisfies(state -> {
            assertThat(state.isEmpty()).isFalse();
            assertThat(state.getCurrentPipelines().keySet()).containsExactlyInAnyOrder(defaultStream.getId(), testStream.getId());
            assertThat(state.getDestinations()).containsExactlyInAnyOrder("indexer");
            assertThat(state.getActiveStreams()).containsExactlyInAnyOrder(defaultStream.getId(), testStream.getId());

            // Asser default stream pipeline
            assertThat(state.getCurrentPipelines().get(defaultStream.getId())).isNotNull().satisfies(pipeline -> {
                assertThat(pipeline.id()).isEqualTo(defaultStream.getId());
                assertThat(pipeline.name()).isEqualTo("Stream Destination Filter: " + defaultStream.getId());
                assertThat(pipeline.stages()).hasSize(1);
                assertThat(pipeline.stages().first()).satisfies(stage -> {
                    assertThat(stage.stage()).isEqualTo(0);
                    assertThat(stage.getRules()).hasSize(1).first().satisfies(rule -> {
                        assertThat(rule.id()).isEqualTo(defaultStreamDestinationFilter.id());
                        assertThat(rule.name()).isEqualTo(ruleTitle(defaultStreamDestinationFilter));
                    });
                    assertThat(stage.ruleReferences())
                            .containsExactlyInAnyOrder(ruleTitle(defaultStreamDestinationFilter));
                });
            });

            // Asser test stream pipeline
            assertThat(state.getCurrentPipelines().get(testStream.getId())).isNotNull().satisfies(pipeline -> {
                assertThat(pipeline.id()).isEqualTo(testStream.getId());
                assertThat(pipeline.name()).isEqualTo("Stream Destination Filter: " + testStream.getId());
                assertThat(pipeline.stages()).hasSize(1);
                assertThat(pipeline.stages().first()).satisfies(stage -> {
                    assertThat(stage.stage()).isEqualTo(0);
                    assertThat(stage.getRules()).hasSize(1).first().satisfies(rule -> {
                        assertThat(rule.id()).isEqualTo(testStreamDestinationFilter.id());
                        assertThat(rule.name()).isEqualTo(ruleTitle(testStreamDestinationFilter));
                    });
                    assertThat(stage.ruleReferences())
                            .containsExactlyInAnyOrder(ruleTitle(testStreamDestinationFilter));
                });
            });

            final var message = messageFactory.createMessage("message", "source", Tools.nowUTC());

            // The message doesn't have a stream, so we shouldn't get any pipelines
            assertThat(state.getPipelinesForMessage(message)).isEmpty();

            // Add the default stream to test that we get the default stream pipeline
            message.addStream(defaultStream);
            assertThat(state.getPipelinesForMessage(message))
                    .hasSize(1)
                    .containsExactlyInAnyOrder(state.getCurrentPipelines().get(defaultStream.getId()));

            // Add the test stream to test that we now get both pipelines
            message.addStream(testStream);
            assertThat(state.getPipelinesForMessage(message))
                    .hasSize(2)
                    .containsAll(state.getCurrentPipelines().values());

            // Remove default stream to test that we now only get the test stream pipeline
            message.removeStream(defaultStream);
            assertThat(state.getPipelinesForMessage(message))
                    .hasSize(1)
                    .containsExactlyInAnyOrder(state.getCurrentPipelines().get(testStream.getId()));
        });
    }