import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
public RedisTestBinder(RedisConnectionFactory connectionFactory) {
	RedisMessageChannelBinder binder = new RedisMessageChannelBinder(connectionFactory);
	GenericApplicationContext context = new GenericApplicationContext();
	context.getBeanFactory().registerSingleton(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
			new DefaultMessageBuilderFactory());
	DefaultHeaderChannelRegistry channelRegistry = new DefaultHeaderChannelRegistry();
	channelRegistry.setReaperDelay(Long.MAX_VALUE);
	ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
	taskScheduler.afterPropertiesSet();
	channelRegistry.setTaskScheduler(taskScheduler);
	context.getBeanFactory().registerSingleton(
			IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
			channelRegistry);
	context.refresh();
	binder.setApplicationContext(context);
	binder.setCodec(new PojoCodec());
	setBinder(binder);
	template = new StringRedisTemplate(connectionFactory);
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
public PubSubTestBinder(PubSub pubSub){
	this.pubSub = pubSub;
	PubSubMessageChannelBinder binder = new PubSubMessageChannelBinder(new PubSubResourceManager(pubSub));
	GenericApplicationContext context = new GenericApplicationContext();
	ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
	scheduler.setPoolSize(1);
	scheduler.afterPropertiesSet();
	context.getBeanFactory().registerSingleton(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, scheduler);
	context.refresh();
	binder.setApplicationContext(context);
	binder.setCodec(new PojoCodec());
	this.setBinder(binder);

}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
@Test
public void testErrorChannelIsBoundWithCorrectContentTypeConverter() {
	final AtomicBoolean received = new AtomicBoolean(false);
	ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
			"--spring.cloud.stream.bindings.error.destination=foo",
			"--spring.cloud.stream.bindings.error.content-type=application/json",
			"--server.port=0");

	MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
			MessageChannel.class);
	MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL,
			MessageChannel.class);

	((SubscribableChannel)errorBridgeChannel).subscribe(new MessageHandler() {
		@Override
		public void handleMessage(Message<?> message) throws MessagingException {
			assertThat(new String((byte[])message.getPayload())).isEqualTo("{\"foo\":\"bar\"}");
			received.set(true);
		}
	});

	Foo foo = new Foo();
	foo.setFoo("bar");

	errorChannel.send(new GenericMessage<>(foo));
	assertThat(received.get()).isTrue();
	applicationContext.close();
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
@Test
public void testErrorChannelForExceptionWhenContentTypeIsSet() {
	final AtomicBoolean received = new AtomicBoolean(false);
	ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
			"--spring.cloud.stream.bindings.error.destination=foo",
			"--spring.cloud.stream.bindings.error.content-type=application/json",
			"--server.port=0");

	MessageChannel errorChannel = applicationContext.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
			MessageChannel.class);
	MessageChannel errorBridgeChannel = applicationContext.getBean(BindingServiceConfiguration.ERROR_BRIDGE_CHANNEL,
			MessageChannel.class);

	((SubscribableChannel)errorBridgeChannel).subscribe(new MessageHandler() {
		@Override
		public void handleMessage(Message<?> message) throws MessagingException {
			String payload = new String((byte[]) message.getPayload());
			assertThat(payload.contains("cause")).isTrue();
			assertThat(payload.contains("stackTrace")).isTrue();
			assertThat(payload.contains("throwing exception")).isTrue();
			received.set(true);
		}
	});

	errorChannel.send(new GenericMessage<>(new Exception("throwing exception")));
	assertThat(received.get()).isTrue();
	applicationContext.close();
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
@Test
public void testErrors() {
	SpringIntegrationChannelBinder binder = createBinder();
	DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
	pollableSource.addInterceptor(new ChannelInterceptorAdapter() {

		@Override
		public Message<?> preSend(Message<?> message, MessageChannel channel) {
			return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
					.copyHeaders(message.getHeaders())
					.build();
		}

	});
	ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
	properties.setMaxAttempts(2);
	properties.setBackOffInitialInterval(0);
	binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
	final CountDownLatch latch = new CountDownLatch(1);
	this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> {
		latch.countDown();
	});
	final AtomicInteger count = new AtomicInteger();
	assertThat(pollableSource.poll(received -> {
		count.incrementAndGet();
		throw new RuntimeException("test recoverer");
	})).isTrue();
	assertThat(count.get()).isEqualTo(2);
	Message<?> lastError = binder.getLastError();
	assertThat(lastError).isNotNull();
	assertThat(((Exception) lastError.getPayload()).getCause().getMessage()).isEqualTo("test recoverer");
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
@Test
public void testErrorsNoRetry() {
	SpringIntegrationChannelBinder binder = createBinder();
	DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
	pollableSource.addInterceptor(new ChannelInterceptorAdapter() {

		@Override
		public Message<?> preSend(Message<?> message, MessageChannel channel) {
			return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
					.copyHeaders(message.getHeaders())
					.build();
		}

	});
	ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
	properties.setMaxAttempts(1);
	binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
	final CountDownLatch latch = new CountDownLatch(1);
	this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> {
		latch.countDown();
	});
	final AtomicInteger count = new AtomicInteger();
	assertThat(pollableSource.poll(received -> {
		count.incrementAndGet();
		throw new RuntimeException("test recoverer");
	})).isTrue();
	assertThat(count.get()).isEqualTo(1);
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
private SpringIntegrationChannelBinder createBinder() {
	SpringIntegrationProvisioner provisioningProvider = new SpringIntegrationProvisioner();
	SpringIntegrationChannelBinder binder = new SpringIntegrationChannelBinder(provisioningProvider);
	this.context.registerBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, PublishSubscribeChannel.class);
	this.context.refresh();
	binder.setApplicationContext(this.context);
	return binder;
}
 

import org.springframework.integration.context.IntegrationContextUtils; //导入依赖的package包/类
@Bean(name = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)
public ConfigurableCompositeMessageConverter configurableCompositeMessageConverter(CompositeMessageConverterFactory factory){
	return new ConfigurableCompositeMessageConverter(factory.getMessageConverterForAllRegistered().getConverters());
}