@Test
public void eventRegistryWithKafka() {
    contextRunner
        .withConfiguration(AutoConfigurations.of(KafkaAutoConfiguration.class))
        .run(context -> {
            assertThat(context)
                .hasSingleBean(KafkaChannelDefinitionProcessor.class)
                .hasBean("kafkaChannelDefinitionProcessor");
            EventRegistryEngine eventRegistryEngine = context.getBean(EventRegistryEngine.class);
            assertThat(eventRegistryEngine).as("Event registry engine").isNotNull();

            IterableAssert<ChannelModelProcessor> channelModelProcessorAssert = assertThat(
                eventRegistryEngine.getEventRegistryEngineConfiguration().getChannelModelProcessors());

            channelModelProcessorAssert
                .hasSize(5);

            channelModelProcessorAssert
                .element(0)
                .isEqualTo(context.getBean("kafkaChannelDefinitionProcessor", ChannelModelProcessor.class));

            channelModelProcessorAssert
                .element(1)
                .isInstanceOf(DelegateExpressionInboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(2)
                .isInstanceOf(DelegateExpressionOutboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(3)
                .isInstanceOf(InboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(4)
                .isInstanceOf(OutboundChannelModelProcessor.class);
        });
}
 

@Test
public void overrideKafkaTopic() throws Exception {
	this.context = new AnnotationConfigApplicationContext();
	environment().setProperty("spring.zipkin.kafka.topic", "zipkin2");
	environment().setProperty("spring.zipkin.sender.type", "kafka");
	this.context.register(PropertyPlaceholderAutoConfiguration.class,
			KafkaAutoConfiguration.class, ZipkinAutoConfiguration.class,
			TraceAutoConfiguration.class);
	this.context.refresh();

	then(this.context.getBean(Sender.class)).isInstanceOf(KafkaSender.class);

	this.context.close();
}
 

@Test
public void canOverrideBySender() throws Exception {
	this.context = new AnnotationConfigApplicationContext();
	environment().setProperty("spring.zipkin.sender.type", "web");
	this.context.register(PropertyPlaceholderAutoConfiguration.class,
			RabbitAutoConfiguration.class, KafkaAutoConfiguration.class,
			ZipkinAutoConfiguration.class, TraceAutoConfiguration.class);
	this.context.refresh();

	then(this.context.getBean(Sender.class).getClass().getName())
			.contains("RestTemplateSender");

	this.context.close();
}
 

@Test
public void canOverrideBySenderAndIsCaseInsensitive() throws Exception {
	this.context = new AnnotationConfigApplicationContext();
	environment().setProperty("spring.zipkin.sender.type", "WEB");
	this.context.register(PropertyPlaceholderAutoConfiguration.class,
			RabbitAutoConfiguration.class, KafkaAutoConfiguration.class,
			ZipkinAutoConfiguration.class, TraceAutoConfiguration.class);
	this.context.refresh();

	then(this.context.getBean(Sender.class).getClass().getName())
			.contains("RestTemplateSender");

	this.context.close();
}
 

@Test
public void rabbitWinsWhenKafkaPresent() throws Exception {
	this.context = new AnnotationConfigApplicationContext();
	environment().setProperty("spring.zipkin.sender.type", "rabbit");
	this.context.register(PropertyPlaceholderAutoConfiguration.class,
			RabbitAutoConfiguration.class, KafkaAutoConfiguration.class,
			ZipkinAutoConfiguration.class, TraceAutoConfiguration.class);
	this.context.refresh();

	then(this.context.getBean(Sender.class)).isInstanceOf(RabbitMQSender.class);

	this.context.close();
}
 

@Test
public void eventRegistryWithJmsRabbitAndKafka() {
    contextRunner
        .withConfiguration(AutoConfigurations.of(
            ActiveMQAutoConfiguration.class,
            JmsAutoConfiguration.class,
            RabbitAutoConfiguration.class,
            KafkaAutoConfiguration.class
        ))
        .run(context -> {
            assertThat(context)
                .hasSingleBean(JmsChannelModelProcessor.class)
                .hasBean("jmsChannelDefinitionProcessor")
                .hasSingleBean(RabbitChannelDefinitionProcessor.class)
                .hasBean("rabbitChannelDefinitionProcessor")
                .hasSingleBean(KafkaChannelDefinitionProcessor.class)
                .hasBean("kafkaChannelDefinitionProcessor");
            EventRegistryEngine eventRegistryEngine = context.getBean(EventRegistryEngine.class);
            assertThat(eventRegistryEngine).as("Event registry engine").isNotNull();

            EventRegistryEngineConfiguration eventRegistryEngineConfiguration = eventRegistryEngine.getEventRegistryEngineConfiguration();
            IterableAssert<ChannelModelProcessor> channelModelProcessorAssert = assertThat(
                eventRegistryEngineConfiguration.getChannelModelProcessors());
            channelModelProcessorAssert
                .hasSize(7)
                .contains(
                    context.getBean("jmsChannelDefinitionProcessor", JmsChannelModelProcessor.class),
                    context.getBean("rabbitChannelDefinitionProcessor", RabbitChannelDefinitionProcessor.class),
                    context.getBean("kafkaChannelDefinitionProcessor", KafkaChannelDefinitionProcessor.class)
                );

            channelModelProcessorAssert
                .element(3)
                .isInstanceOf(DelegateExpressionInboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(4)
                .isInstanceOf(DelegateExpressionOutboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(5)
                .isInstanceOf(InboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(6)
                .isInstanceOf(OutboundChannelModelProcessor.class);
        });
}
 

@Test
public void eventRegistryWithCustomDefinitionProcessors() {
    contextRunner
        .withConfiguration(AutoConfigurations.of(
            ActiveMQAutoConfiguration.class,
            JmsAutoConfiguration.class,
            RabbitAutoConfiguration.class,
            KafkaAutoConfiguration.class
        ))
        .withUserConfiguration(CustomChannelDefinitionProcessorsConfiguration.class)
        .run(context -> {
            assertThat(context)
                .doesNotHaveBean(JmsChannelModelProcessor.class)
                .hasBean("jmsChannelDefinitionProcessor")
                .doesNotHaveBean(RabbitChannelDefinitionProcessor.class)
                .hasBean("rabbitChannelDefinitionProcessor")
                .doesNotHaveBean(KafkaChannelDefinitionProcessor.class)
                .hasBean("kafkaChannelDefinitionProcessor")
                .hasBean("customChannelDefinitionProcessor");
            EventRegistryEngine eventRegistryEngine = context.getBean(EventRegistryEngine.class);
            assertThat(eventRegistryEngine).as("Event registry engine").isNotNull();

            EventRegistryEngineConfiguration eventRegistryEngineConfiguration = eventRegistryEngine.getEventRegistryEngineConfiguration();
            IterableAssert<ChannelModelProcessor> channelModelProcessorAssert = assertThat(
                eventRegistryEngineConfiguration.getChannelModelProcessors());
            channelModelProcessorAssert
                .hasSize(8);

            channelModelProcessorAssert
                .element(0)
                .isEqualTo(context.getBean("customChannelDefinitionProcessor", ChannelModelProcessor.class));

            channelModelProcessorAssert
                .element(1)
                .isEqualTo(context.getBean("rabbitChannelDefinitionProcessor", ChannelModelProcessor.class));

            channelModelProcessorAssert
                .element(2)
                .isEqualTo(context.getBean("jmsChannelDefinitionProcessor", ChannelModelProcessor.class));

        channelModelProcessorAssert
                .element(3)
                .isEqualTo(context.getBean("kafkaChannelDefinitionProcessor", ChannelModelProcessor.class));

            channelModelProcessorAssert
                .element(4)
                .isInstanceOf(DelegateExpressionInboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(5)
                .isInstanceOf(DelegateExpressionOutboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(6)
                .isInstanceOf(InboundChannelModelProcessor.class);

            channelModelProcessorAssert
                .element(7)
                .isInstanceOf(OutboundChannelModelProcessor.class);
        });
}