@@ -102,16 +102,16 @@ public class BinderAwareChannelResolver
|
||||
}
|
||||
else {
|
||||
channel = this.bindingTargetFactory.createOutput(channelName);
|
||||
ProducerProperties producerProperties = bindingServiceProperties
|
||||
.getProducerProperties(channelName);
|
||||
if (this.newBindingCallback != null) {
|
||||
ProducerProperties producerProperties = bindingServiceProperties
|
||||
.getProducerProperties(channelName);
|
||||
Object extendedProducerProperties = this.bindingService
|
||||
.getExtendedProducerProperties(channel, channelName);
|
||||
this.newBindingCallback.configure(channelName, channel,
|
||||
producerProperties, extendedProducerProperties);
|
||||
bindingServiceProperties.updateProducerProperties(channelName,
|
||||
producerProperties);
|
||||
}
|
||||
bindingServiceProperties.updateProducerProperties(channelName,
|
||||
producerProperties);
|
||||
this.beanFactory.registerSingleton(channelName, channel);
|
||||
channel = (MessageChannel) this.beanFactory.initializeBean(channel,
|
||||
channelName);
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
/*
|
||||
* Copyright 2019-2019 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.stream.function;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
|
||||
import org.springframework.cloud.stream.binder.test.InputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.OutputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
public class DynamicDestinationFunctionTests {
|
||||
|
||||
@After
|
||||
public void after() {
|
||||
System.clearProperty("spring.cloud.stream.function.definition");
|
||||
System.clearProperty("spring.cloud.function.definition");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testEmptyConfiguration() {
|
||||
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(SampleConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.bindings.fooDestination.producer.partitionKeyExtractorName=keyExtractor")) {
|
||||
InputDestination input = context.getBean(InputDestination.class);
|
||||
input.send(new GenericMessage<String>("fooDestination"));
|
||||
|
||||
BindingServiceProperties serviceProperties = context.getBean(BindingServiceProperties.class);
|
||||
assertThat("keyExtractor").isEqualTo(
|
||||
serviceProperties.getProducerProperties("fooDestination").getPartitionKeyExtractorName());
|
||||
|
||||
OutputDestination output = context.getBean(OutputDestination.class);
|
||||
assertThat(output.receive(1000).getPayload()).isEqualTo("fooDestination".getBytes());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class SampleConfiguration {
|
||||
|
||||
@Autowired
|
||||
private BinderAwareChannelResolver resolver;
|
||||
|
||||
@Bean
|
||||
public PartitionKeyExtractorStrategy keyExtractor() {
|
||||
return new PartitionKeyExtractorStrategy() {
|
||||
|
||||
@Override
|
||||
public Object extractKey(Message<?> message) {
|
||||
return 0;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Consumer<String> cons() {
|
||||
return value -> {
|
||||
System.out.println("Sending to " + value);
|
||||
resolver.resolveDestination(value).send(new GenericMessage<String>(value));
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user