CODEWITHSUNDEEP
javaspring-cloud-stream
cherry-peng
cherry-peng

Reputation: 3

I used spring-cloud-stream can not send message, but can receive message

StreamInput.java

public interface StreamInput {
    String QUEUE = "cherryQueue";
    @Input(StreamInput.QUEUE)
    SubscribableChannel input();
}

StreamOutput.java

public interface StreamOutput {
    @Output(StreamInput.QUEUE)
    MessageChannel output();
}

StreamReceiver.java

@Slf4j
@Component
@EnableBinding(StreamInput.class)
public class StreamReceiver {
    @StreamListener(StreamInput.QUEUE)
    public void process(Object message){
    log.info("receive message:{}",message);
    log.info("--------------------------------------------------");
    }
}

the springboot project start up OK，and the queue create success： enter image description here

======================================================================== but！I use the StreamOutput, it doesn't work .

this is Test case:

@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@EnableBinding(StreamOutput.class)
public class MessageTest {

    @Autowired
    private StreamOutput streamOutput;

    @Test
    public void sentMessageTest() {
        streamOutput.output()
                .send(MessageBuilder.withPayload("now time is :" + new Date()).build());
        log.info("the message send OK !!!");
    }

}

this is the Exception info :

java.lang.IllegalStateException: Failed to load ApplicationContext

    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:125)
    at org.springframework.test.context.support.DefaultTestContext.getApplicationContext(DefaultTestContext.java:108)
    at org.springframework.test.context.web.ServletTestExecutionListener.setUpRequestContextIfNecessary(ServletTestExecutionListener.java:190)
    at org.springframework.test.context.web.ServletTestExecutionListener.prepareTestInstance(ServletTestExecutionListener.java:132)
    at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:246)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:227)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner$1.runReflectiveCall(SpringJUnit4ClassRunner.java:289)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:291)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:246)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
    at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
    at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
    at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
    at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
    at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null
    at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:64)
    at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerInputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:47)
    at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions$0(BindingBeanDefinitionRegistryUtils.java:80)
    at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:562)
    at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:541)
    at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:76)
    at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:45)
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars$1(ConfigurationClassBeanDefinitionReader.java:358)
    at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684)
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:357)
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145)
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117)
    at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:328)
    at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:233)
    at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:273)
    at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:93)
    at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:694)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:532)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:759)
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:395)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:327)
    at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:139)
    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContextInternal(DefaultCacheAwareContextLoaderDelegate.java:99)
    at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:117)
    ... 24 more

Upvotes: 0

Views: 550

Answers (2)

tbr
tbr

Reputation: 11

Why this scenario is unusual, I do plan to develop two microservices initialy as one app to speedup the development and use messages to communicate.

Everything was working with pure Rabbit, now it stopped when I switched to Spring Cloud Stream.

I do have also another 'real' Sink in another microservice.

Defining Sink & Source with different 'bean names' and overriden destinations still does not work because I suspect Spring forwards messages from Source using same thread (DirectChannel) to the internal sink (skipping rabbit!).

Upvotes: 0

Gary Russell
Gary Russell

Reputation: 174799

Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null

The error is quite clear, I think.

@Output(StreamInput.QUEUE)

You can't use the same channel name for 2 channels.

If you wish to produce and consume messages from the same destination in the same application (which would be a little unusual) you must use 2 channels with the bindings configured to reference the same destination.

Upvotes: 1

Related Questions