import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@Ignore("Kinesalite doesn't support updateShardCount. Test only against real AWS Kinesis")
public void testPartitionCountIncreasedIfAutoAddPartitionsSet() {
	KinesisBinderConfigurationProperties configurationProperties = new KinesisBinderConfigurationProperties();

	String stream = "existing" + System.currentTimeMillis();

	AmazonKinesisAsync amazonKinesis = localKinesisResource.getResource();
	amazonKinesis.createStream(stream, 1);

	List<Shard> shards = describeStream(stream);

	assertThat(shards.size()).isEqualTo(1);

	configurationProperties.setMinShardCount(6);
	configurationProperties.setAutoAddShards(true);
	KinesisTestBinder binder = getBinder(configurationProperties);

	ExtendedConsumerProperties<KinesisConsumerProperties> consumerProperties = createConsumerProperties();
	Binding<?> binding = binder.bindConsumer(stream, "test", new NullChannel(), consumerProperties);
	binding.unbind();

	shards = describeStream(stream);

	assertThat(shards.size()).isEqualTo(6);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
private Binding<MessageChannel> doRegisterConsumer(String bindingName, String group, String channelName, MessageChannel moduleInputChannel,
		MessageProducerSupport adapter, final ConsumerProperties properties) {
	DirectChannel bridgeToModuleChannel = new DirectChannel();
	bridgeToModuleChannel.setBeanFactory(this.getBeanFactory());
	bridgeToModuleChannel.setBeanName(channelName + ".bridge");
	MessageChannel bridgeInputChannel = addRetryIfNeeded(channelName, bridgeToModuleChannel, properties);
	adapter.setOutputChannel(bridgeInputChannel);
	adapter.setBeanName("inbound." + channelName);
	adapter.afterPropertiesSet();
	DefaultBinding<MessageChannel> consumerBinding = new DefaultBinding<MessageChannel>(bindingName, group, moduleInputChannel, adapter) {

		@Override
		protected void afterUnbind() {
			String key = RedisMessageChannelBinder.CONSUMER_GROUPS_KEY_PREFIX + getName();
			RedisMessageChannelBinder.this.redisOperations.boundZSetOps(key).incrementScore(getGroup(), -1);
		}
	};
	ReceivingHandler convertingBridge = new ReceivingHandler(properties);
	convertingBridge.setOutputChannel(moduleInputChannel);
	convertingBridge.setBeanName(channelName + ".bridge.handler");
	convertingBridge.afterPropertiesSet();
	bridgeToModuleChannel.subscribe(convertingBridge);
	this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + bindingName).incrementScore(group, 1);
	adapter.start();
	return consumerBinding;
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
private Binding<MessageChannel> doRegisterProducer(final String name, MessageChannel moduleOutputChannel,
		ProducerProperties properties) {
	Assert.isInstanceOf(SubscribableChannel.class, moduleOutputChannel);
	MessageHandler handler = new SendingHandler(name, properties);
	EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) moduleOutputChannel, handler);
	consumer.setBeanFactory(this.getBeanFactory());
	consumer.setBeanName("outbound." + name);
	consumer.afterPropertiesSet();
	DefaultBinding<MessageChannel> producerBinding =
			new DefaultBinding<>(name, null, moduleOutputChannel, consumer);
	String[] requiredGroups = properties.getRequiredGroups();
	if (!ObjectUtils.isEmpty(requiredGroups)) {
		for (String group : requiredGroups) {
			this.redisOperations.boundZSetOps(CONSUMER_GROUPS_KEY_PREFIX + name).incrementScore(group, 1);
		}
	}
	consumer.start();
	return producerBinding;
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
public void testRetryFail() {
	RedisTestBinder binder = getBinder();
	DirectChannel channel = new DirectChannel();
	binder.bindProducer("retry.0", channel, createProducerProperties());
	ConsumerProperties consumerProperties = new ConsumerProperties();
	consumerProperties.setMaxAttempts(2);
	consumerProperties.setBackOffInitialInterval(100);
	consumerProperties.setBackOffMultiplier(1.0);
	Binding<MessageChannel> consumerBinding = binder.bindConsumer("retry.0", "test", new DirectChannel(), consumerProperties); // no subscriber
	channel.send(new GenericMessage<>("foo"));
	RedisTemplate<String, Object> template = createTemplate();
	Object rightPop = template.boundListOps("ERRORS:retry.0.test").rightPop(5, TimeUnit.SECONDS);
	assertNotNull(rightPop);
	assertThat(new String((byte[]) rightPop), containsString("foo"));
	consumerBinding.unbind();
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@Override
public void testSendAndReceive() throws Exception {
	RedisTestBinder binder = getBinder();
	DirectChannel moduleOutputChannel = new DirectChannel();
	QueueChannel moduleInputChannel = new QueueChannel();
	ProducerProperties producerProperties = createProducerProperties();
	producerProperties.setHeaderMode(HeaderMode.raw);
	Binding<MessageChannel> producerBinding = binder.bindProducer("foo.0", moduleOutputChannel, producerProperties);
	ConsumerProperties consumerProperties = createConsumerProperties();
	consumerProperties.setHeaderMode(HeaderMode.raw);
	Binding<MessageChannel> consumerBinding = binder.bindConsumer("foo.0", "test", moduleInputChannel, consumerProperties);
	Message<?> message = MessageBuilder.withPayload("foo".getBytes()).build();
	// Let the consumer actually bind to the producer before sending a msg
	binderBindUnbindLatency();
	moduleOutputChannel.send(message);
	Message<?> inbound = receive(moduleInputChannel);
	assertNotNull(inbound);
	assertEquals("foo", new String((byte[])inbound.getPayload()));
	producerBinding.unbind();
	consumerBinding.unbind();
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Throwable {
	KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

	String testTopicName = "existing" + System.currentTimeMillis();
	invokeCreateTopic(testTopicName, 5, 1);
	configurationProperties.setAutoCreateTopics(false);
	Binder binder = getBinder(configurationProperties);

	ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();

	DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));
	Binding<MessageChannel> binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties);
	binding.unbind();
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Throwable {
	KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

	String testTopicName = "existing" + System.currentTimeMillis();
	configurationProperties.setMinPartitionCount(6);
	configurationProperties.setAutoAddPartitions(true);
	Binder binder = getBinder(configurationProperties);
	ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
	DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));

	Binding<?> binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties);
	binding.unbind();
	assertThat(invokePartitionSize(testTopicName)).isEqualTo(6);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled() throws Throwable {
	KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

	String testTopicName = "existing" + System.currentTimeMillis();
	invokeCreateTopic(testTopicName, 1, 1);
	configurationProperties.setAutoAddPartitions(false);
	Binder binder = getBinder(configurationProperties);
	GenericApplicationContext context = new GenericApplicationContext();
	context.refresh();

	ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();

	DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));

	// this consumer must consume from partition 2
	consumerProperties.setInstanceCount(3);
	consumerProperties.setInstanceIndex(2);
	Binding binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties);
	binding.unbind();
	assertThat(invokePartitionSize(testTopicName)).isEqualTo(1);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled() throws Throwable {
	KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

	String testTopicName = "existing" + System.currentTimeMillis();
	invokeCreateTopic(testTopicName, 1, 1);
	configurationProperties.setAutoAddPartitions(false);
	Binder binder = getBinder(configurationProperties);
	GenericApplicationContext context = new GenericApplicationContext();
	context.refresh();

	ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
	DirectChannel output = createBindableChannel("output", createConsumerBindingProperties(consumerProperties));
	// this consumer must consume from partition 2
	consumerProperties.setInstanceCount(3);
	consumerProperties.setInstanceIndex(2);
	consumerProperties.getExtension().setAutoRebalanceEnabled(false);
	expectedProvisioningException.expect(ProvisioningException.class);
	expectedProvisioningException
			.expectMessage("The number of expected partitions was: 3, but 1 has been found instead");
	Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
	if (binding != null) {
		binding.unbind();
	}
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testPartitionCountNotReduced() throws Throwable {
	String testTopicName = "existing" + System.currentTimeMillis();

	KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

	invokeCreateTopic(testTopicName, 6, 1);
	configurationProperties.setAutoAddPartitions(true);
	Binder binder = getBinder(configurationProperties);
	GenericApplicationContext context = new GenericApplicationContext();
	context.refresh();

	ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
	DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));

	Binding<?> binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties);
	binding.unbind();

	assertThat(partitionSize(testTopicName)).isEqualTo(6);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void testConsumerDefaultDeserializer() throws Throwable {
	Binding<?> binding = null;
	try {
		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
		String testTopicName = "existing" + System.currentTimeMillis();
		invokeCreateTopic(testTopicName, 5, 1);
		configurationProperties.setAutoCreateTopics(false);
		Binder binder = getBinder(configurationProperties);

		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
		DirectChannel input = createBindableChannel("input", createConsumerBindingProperties(consumerProperties));

		binding = binder.bindConsumer(testTopicName, "test", input, consumerProperties);
		DirectFieldAccessor consumerAccessor = new DirectFieldAccessor(getKafkaConsumer(binding));
		assertTrue(consumerAccessor.getPropertyValue("keyDeserializer") instanceof ByteArrayDeserializer);
		assertTrue(consumerAccessor.getPropertyValue("valueDeserializer") instanceof ByteArrayDeserializer);
	}
	finally {
		if (binding != null) {
			binding.unbind();
		}
	}
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
public void shouldPublishMessageWithBytePayload() {

    // given
    DirectChannel output = new DirectChannel();

    ArgumentCaptor<URI> uriCaptor = ArgumentCaptor.forClass(URI.class);
    ArgumentCaptor<HermesMessage> messageCaptor = ArgumentCaptor.forClass(HermesMessage.class);

    // when
    Binding<MessageChannel> binding = binder.bindProducer(
            OUTPUT_NAME, output, new ExtendedProducerProperties<>(new HermesProducerProperties()));

    // then
    output.send(new GenericMessage<>(MESSAGE, json()));
    verify(hermesSender).send(uriCaptor.capture(), messageCaptor.capture());

    assertEquals("http://localhost:8080/topics/topic", uriCaptor.getValue().toString());
    assertArrayEquals(MESSAGE.getBytes(), messageCaptor.getValue().getBody());

    binding.unbind();
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
public void shouldPublishMessageWithError() {

    // given
    reset(hermesSender);
    final HermesResponse response = HermesResponseBuilder.hermesResponse()
            .withHttpStatus(500)
            .build();

    when(hermesSender.send(any(URI.class), any(HermesMessage.class)))
            .thenReturn(CompletableFuture.completedFuture(response));

    DirectChannel output = new DirectChannel();

    // when
    Binding<MessageChannel> binding = binder.bindProducer(
            OUTPUT_NAME, output, new ExtendedProducerProperties<>(new HermesProducerProperties()));

    // then
    output.send(new GenericMessage<>(MESSAGE, json()));
    verify(hermesSender, times(4)).send(any(URI.class), any(HermesMessage.class));
    binding.unbind();
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Override
public Binding<MessageChannel> bindConsumer(String name, String group, MessageChannel moduleInputChannel,
		ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
	if (group != null) {
		if (properties.getExtension().isQueueNameGroupOnly()) {
			this.queues.add(properties.getExtension().getPrefix() + group);
		}
		else {
			this.queues.add(properties.getExtension().getPrefix() + name + ("." + group));
		}
	}
	this.exchanges.add(properties.getExtension().getPrefix() + name);
	this.prefixes.add(properties.getExtension().getPrefix());
	deadLetters(properties.getExtension());
	return super.bindConsumer(name, group, moduleInputChannel, properties);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Override
public Binding<MessageChannel> bindProducer(String name, MessageChannel moduleOutputChannel,
		ExtendedProducerProperties<RabbitProducerProperties> properties) {
	this.queues.add(properties.getExtension().getPrefix() + name + ".default");
	this.exchanges.add(properties.getExtension().getPrefix() + name);
	if (properties.getRequiredGroups() != null) {
		for (String group : properties.getRequiredGroups()) {
			if (properties.getExtension().isQueueNameGroupOnly()) {
				this.queues.add(properties.getExtension().getPrefix() + group);
			}
			else {
				this.queues.add(properties.getExtension().getPrefix() + name + "." + group);
			}
		}
	}
	this.prefixes.add(properties.getExtension().getPrefix());
	deadLetters(properties.getExtension());
	return super.bindProducer(name, moduleOutputChannel, properties);
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
public <T> Binding<T> doBindConsumer(T input, String inputName, Binder<T, ConsumerProperties, ?> binder,
		ConsumerProperties consumerProperties, String target) {
	if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
		return binder.bindConsumer(target,
				this.bindingServiceProperties.getGroup(inputName), input,
				consumerProperties);
	}
	else {
		try {
			return binder.bindConsumer(target,
					this.bindingServiceProperties.getGroup(inputName), input,
					consumerProperties);
		}
		catch (RuntimeException e) {
			LateBinding<T> late = new LateBinding<T>();
			rescheduleConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
			return late;
		}
	}
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@SuppressWarnings({ "rawtypes", "unchecked" })
public <T> Binding<T> doBindPollableConsumer(T input, String inputName, Binder<T, ConsumerProperties, ?> binder,
		ConsumerProperties consumerProperties, String target) {
	if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
		return ((PollableConsumerBinder) binder).bindPollableConsumer(target,
				this.bindingServiceProperties.getGroup(inputName), (PollableSource) input,
				consumerProperties);
	}
	else {
		try {
			return ((PollableConsumerBinder) binder).bindPollableConsumer(target,
					this.bindingServiceProperties.getGroup(inputName), (PollableSource) input,
					consumerProperties);
		}
		catch (RuntimeException e) {
			LateBinding<T> late = new LateBinding<T>();
			reschedulePollableConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
			return late;
		}
	}
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@SuppressWarnings({ "unchecked", "rawtypes" })
public <T> Binding<T> bindProducer(T output, String outputName) {
	String bindingTarget = this.bindingServiceProperties
			.getBindingDestination(outputName);
	Binder<T, ?, ProducerProperties> binder = (Binder<T, ?, ProducerProperties>) getBinder(
			outputName, output.getClass());
	ProducerProperties producerProperties = this.bindingServiceProperties
			.getProducerProperties(outputName);
	if (binder instanceof ExtendedPropertiesBinder) {
		Object extension = ((ExtendedPropertiesBinder) binder)
				.getExtendedProducerProperties(outputName);
		ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>(
				extension);
		BeanUtils.copyProperties(producerProperties, extendedProducerProperties);
		producerProperties = extendedProducerProperties;
	}
	validate(producerProperties);
	Binding<T> binding = doBindProducer(output, bindingTarget, binder, producerProperties);
	this.producerBindings.put(outputName, binding);
	return binding;
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
public <T> Binding<T> doBindProducer(T output, String bindingTarget, Binder<T, ?, ProducerProperties> binder,
		ProducerProperties producerProperties) {
	if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
		return binder.bindProducer(bindingTarget, output, producerProperties);
	}
	else {
		try {
			return binder.bindProducer(bindingTarget, output, producerProperties);
		}
		catch (RuntimeException e) {
			LateBinding<T> late = new LateBinding<T>();
			rescheduleProducerBinding(output, bindingTarget, binder, producerProperties, late, e);
			return late;
		}
	}
}
 

import org.springframework.cloud.stream.binder.Binding; //导入依赖的package包/类
@Test
public void testAutoCreateStreamForNonExistingStream() throws Exception {
	KinesisTestBinder binder = getBinder();
	DirectChannel output = createBindableChannel("output", new BindingProperties());
	ExtendedConsumerProperties<KinesisConsumerProperties> consumerProperties = createConsumerProperties();
	Date testDate = new Date();
	consumerProperties.getExtension()
			.setShardIteratorType(ShardIteratorType.AT_TIMESTAMP.name() + ":" + testDate.getTime());
	String testStreamName = "nonexisting" + System.currentTimeMillis();
	Binding<?> binding = binder.bindConsumer(testStreamName, "test", output, consumerProperties);
	binding.unbind();

	DescribeStreamResult streamResult = localKinesisResource.getResource().describeStream(testStreamName);
	String createdStreamName = streamResult.getStreamDescription().getStreamName();
	int createdShards = streamResult.getStreamDescription().getShards().size();
	String createdStreamStatus = streamResult.getStreamDescription().getStreamStatus();

	assertThat(createdStreamName).isEqualTo(testStreamName);
	assertThat(createdShards)
			.isEqualTo(consumerProperties.getInstanceCount() * consumerProperties.getConcurrency());
	assertThat(createdStreamStatus).isEqualTo(StreamStatus.ACTIVE.toString());

	KinesisShardOffset shardOffset =
			TestUtils.getPropertyValue(binding, "lifecycle.streamInitialSequence", KinesisShardOffset.class);
	assertThat(shardOffset.getIteratorType()).isEqualTo(ShardIteratorType.AT_TIMESTAMP);
	assertThat(shardOffset.getTimestamp()).isEqualTo(testDate);
}