org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration#zipkin2.reporter.amqp.RabbitMQSender源码实例Demo

/**
 * Build a new {@link HttpTracing} instance for interfacing with Zipkin
 *
 * @param environment Environment
 * @return Brave instance
 */
@Override
public Optional<HttpTracing> build(@NotNull final Environment environment) {
  if (!isEnabled()) {
    LOGGER.warn("Zipkin tracing is disabled");
    return Optional.empty();
  }

  final RabbitMQSender sender =
      RabbitMQSender.newBuilder()
          .addresses(addresses)
          .queue(queue)
          .connectionTimeout((int) connectionTimeout.toMilliseconds())
          .username(username)
          .password(password)
          .virtualHost(virtualHost)
          .build();

  LOGGER.info("Sending spans to RabbitMQ queue \"{}\" at: {}", queue, addresses);

  return buildTracing(environment, sender);
}
 

/**
 * Create reporter.
 */
private AsyncReporter<Span> setupReporter(Encoding encoding) {
    var sender = RabbitMQSender.newBuilder()
            .username("guest")
            .username("guest")
            .virtualHost("/")
            .encoding(encoding)
            .queue("zipkin")
            .addresses(rabbitMqContainer.getContainerIpAddress() + ":" + rabbitMqContainer.getFirstMappedPort())
            .build();
    return AsyncReporter.create(sender);
}
 

@Override protected RabbitMQSender createInstance() {
  RabbitMQSender.Builder builder = RabbitMQSender.newBuilder();
  if (addresses != null) builder.addresses(addresses);
  if (encoding != null) builder.encoding(encoding);
  if (queue != null) builder.queue(queue);
  if (connectionTimeout != null) builder.connectionTimeout(connectionTimeout);
  if (virtualHost != null) builder.virtualHost(virtualHost);
  if (username != null) builder.username(username);
  if (password != null) builder.password(password);
  if (messageMaxBytes != null) builder.messageMaxBytes(messageMaxBytes);
  return builder.build();
}
 

@Test public void addresses() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("addresses")
      .isEqualTo(asList(new Address("localhost")));
}
 

@Test public void queue() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"queue\" value=\"zipkin2\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("queue")
      .isEqualTo("zipkin2");
}
 

@Test public void connectionTimeout() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"connectionTimeout\" value=\"0\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("connectionFactory.connectionTimeout")
      .isEqualTo(0);
}
 

@Test public void virtualHost() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"virtualHost\" value=\"zipkin3\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("connectionFactory.virtualHost")
      .isEqualTo("zipkin3");
}
 

@Test public void usernamePassword() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"username\" value=\"foo\"/>\n"
      + "  <property name=\"password\" value=\"bar\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("connectionFactory.username", "connectionFactory.password")
      .isEqualTo(asList("foo", "bar"));
}
 

@Test public void messageMaxBytes() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"messageMaxBytes\" value=\"1024\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("messageMaxBytes")
      .isEqualTo(1024);
}
 

@Test public void encoding() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "  <property name=\"encoding\" value=\"PROTO3\"/>\n"
      + "</bean>"
  );

  assertThat(context.getBean("sender", RabbitMQSender.class))
      .extracting("encoding")
      .isEqualTo(Encoding.PROTO3);
}
 

@Test(expected = IllegalStateException.class) public void close_closesSender() {
  context = new XmlBeans(""
      + "<bean id=\"sender\" class=\"zipkin2.reporter.beans.RabbitMQSenderFactoryBean\">\n"
      + "  <property name=\"addresses\" value=\"localhost\"/>\n"
      + "</bean>"
  );

  RabbitMQSender sender = context.getBean("sender", RabbitMQSender.class);
  context.close();

  sender.sendSpans(asList(new byte[] {'{', '}'}));
}
 

@Bean(ZipkinAutoConfiguration.SENDER_BEAN_NAME)
Sender rabbitSender(CachingConnectionFactory connectionFactory,
		RabbitProperties config) {
	String addresses = StringUtils.hasText(this.addresses) ? this.addresses
			: config.determineAddresses();
	return RabbitMQSender.newBuilder()
			.connectionFactory(connectionFactory.getRabbitConnectionFactory())
			.queue(this.queue).addresses(addresses).build();
}
 

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

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

	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();
}
 

@Override public Class<? extends RabbitMQSender> getObjectType() {
  return RabbitMQSender.class;
}
 

@Override protected void destroyInstance(Object instance) throws IOException {
  ((RabbitMQSender) instance).close();
}