Announcement Announcement Module
Collapse
No announcement yet.
Publish Subscribe channel Page Title Module
Move Remove Collapse
X
Conversation Detail Module
Collapse
  • Filter
  • Time
  • Show
Clear All
new posts

  • Publish Subscribe channel

    I just started learning Spring integration.

    What I understand is Publish Subscribe channel is used to send message to more than one receiver who have subscribed to channel.

    So I tried following.

    Code:
    <int:publish-subscribe-channel id="pubsub" />
    
    <int:channel id="test2" />
    <int:channel id="test3" />
    
    <int:recipient-list-router id="customRouter" input-channel="pubsub">
    		<int:recipient channel="test2" />
    		<int:recipient channel="test3" />
    </int:recipient-list-router>
    
    <int:poller default="true" fixed-rate="100"></int:poller>


    Then I wrote following code.

    Code:
    public class Test1 {
    	public static void main(String args[]) {
    		try {
    			ApplicationContext context = new ClassPathXmlApplicationContext("classpath:beans1.xml");
    			SubscribableChannel channel1 = (SubscribableChannel)context.getBean("pubsub");
    			System.out.println(channel1.send(MessageBuilder.withPayload("Hello there").build()));
    			
    			PollableChannel channel2 = (PollableChannel)context.getBean("test2");
    			System.out.println(channel2.receive());
    		}
    		catch(Exception e) {
    			System.out.println(e);
    		}
    	}
    }

    But I get following exception.

    org.springframework.integration.MessageDeliveryExc eption: failed to send Message to channel 'test2'

  • #2
    If you want the 'test2' and 'test3' channels to be PollableChannels (to buffer messages that are received explicitly by pollers), then add a <queue> sub-element to each. That should do the trick.

    Comment


    • #3
      I would also suggest to go through Spring Integration samples http://blog.springsource.com/2010/09...ation-samples/
      Those should answer most of the 'startup' questions.

      Comment


      • #4
        Thanks for reply but it did not work.

        I added following queue element.




        Code:
        <int:channel id="test2">
                 <int:queue capacity="10"/>
        </int:channel>
        <int:channel id="test3">
                 <int:queue capacity="10"/>
        </int:channel>
        I got rid of exception now. But I could not send message to pubsub channel at all.

        When I printed following

        Code:
        System.out.println(channel1.send(MessageBuilder.withPayload("Hello there").build()));
        output was : false

        which means it did not send message to pubsub channel.

        Comment


        • #5
          Why do you have a pub-sub channel and a recipient list router?

          There is only one subscriber to the channel (the router).

          Normally, one would have a pub-sub channel with multiple subscribers.

          Code:
          <publish-subscribe-channel id="psc"/>
          
          <service-activator id="sub1" input-channel="psc" ... />
          
          <service-activator id="sub2" input-channel="psc" .../>
          This would run sub1 first, then sub2.

          If you want them to run in parallel, add a task executor to psc.

          Comment


          • #6
            Originally posted by oleg.zhurakousky View Post
            I would also suggest to go through Spring Integration samples http://blog.springsource.com/2010/09...ation-samples/
            Those should answer most of the 'startup' questions.
            You know what? I did what you asked and completely wasted my time. You have absolutely not a single example of publish-subscribe-channel. Not a single one. Why are you directing readers to non-existent help somewhere else? It's irritating.

            I'm trying simple pub sub and it appears that the second channel is not getting the message.

            Any ideas why the second service-activator is not getting the message? No, this is not covered in the samples.


            <int:json-to-object-transformer input-channel="fromAmqpOrdersChannel"
            output-channel="toTradingServiceChannel"
            type="com.reditrader.domain.Order"/>

            <intublish-subscribe-channel id="toTradingServiceChannel" />

            <int:service-activator input-channel="toTradingServiceChannel"
            output-channel="nullChannel"
            ref="tradingServiceImpl" method="saveOrder"/>

            <int:logging-channel-adapter channel="toTradingServiceChannel" expression="'1:' + payload" logger-name="test.pubsub.1"/>

            <int:service-activator input-channel="toTradingServiceChannel"
            output-channel="nullChannel"
            ref="rediOrderApiImpl" method="sendOrder"/>

            Comment


            • #7
              There is an example that shows publish-subscribe-channel https://github.com/SpringSource/spri...sting-examples , but i guess we need to have a more trivial one. Feel free to raise a JIRA issue https://jira.springsource.org/browse/INTSAMPLES
              With reagard to your configuration I am suspecting (since you showing part of your configuration) that json-to-object-transformer is not receiving any messages thus not sending any and therefore non of your subscribers receive a message. I just tried to use your config and send a message directly to 'toTradingServiceChannel' and it works just fine

              Comment


              • #8
                Also, you can take any example that has a DirectChannel and substitute it with PublishSubscribeChannel. For example; in 'helloworld sample you can substitute
                Code:
                <channel id="inputChannel"/>
                with
                Code:
                <publish-subscribe-channel id="inputChannel"/>
                First thing you'll notice is tat everything will work as before. But you can now start adding more subscribers.
                And of course Reference Manual has plenty of information http://static.springsource.org/sprin...ingle/#channel

                Comment


                • #9
                  Also, you should be aware that, by default, the subscribers are invoked serially on the same thread; if the first subscriber throws an exception, the second subscriber won't get the message. You can change this default behavior by setting...

                  Code:
                  ignore-failures="true"
                  ...or by adding a task executor so each subscriber is called on a different thread.

                  Comment


                  • #10
                    @owebeewan, hopefully you got all the information you need from the recent replies. I wanted to point out a few things...

                    First, the original post and Oleg's reply to that was over a year old. We have made a lot of changes to the samples and moved them to GitHub since then: https://github.com/SpringSource/spri...ration-samples

                    Second, I agree that we really should have a very simple 'pubsub' example in there, so please do raise a JIRA issue for that: https://jira.springsource.org/browse/INTSAMPLES

                    Finally, we try really hard to keep this forum as low tension as possible, so next time please just ask a question here (e.g. for a more specific pointer) rather than wasting too much of your own time and then posting with such an accusatory tone. Everyone who spends time in this forum - whether a developer on the Spring Integration team or a member of the community - is just trying to help.

                    Thanks,
                    Mark

                    Comment


                    • #11
                      Originally posted by Gary Russell View Post
                      Also, you should be aware that, by default, the subscribers are invoked serially on the same thread; if the first subscriber throws an exception, the second subscriber won't get the message. You can change this default behavior by setting...

                      Code:
                      ignore-failures="true"
                      ...or by adding a task executor so each subscriber is called on a different thread.

                      Gary! You win! This is awesome. Your friend Brian D. wrote (or supervised the writing of) the first module and I presupposed that it would never have an error in it. We reviewed it together today and after much time and investigation, voila! ignore-failures=true was the answer. It was throwing an error deep in the bowels of the first program and was never being reported anywhere. Brian and I contemplated how you could have possibly inferred this, but, thank you for this precision. Lesson to always use error queues also.

                      Comment


                      • #12
                        Originally posted by Mark Fisher View Post
                        Finally, we try really hard to keep this forum as low tension as possible, so next time please just ask a question here (e.g. for a more specific pointer) rather than wasting too much of your own time and then posting with such an accusatory tone. Everyone who spends time in this forum - whether a developer on the Spring Integration team or a member of the community - is just trying to help.

                        Thanks,
                        Mark
                        Thanks Mark. It is very uncharacteristic of me as I am naturally mellow and a generally all-around cool laid-back guy. I started working on this at 1pm and it was 3am when I posted this after following the "samples" trail. Kudos to all of you for your hard work. Brian D. says that he loves all of you. If you're good enough for Brian, then you're way good for me! Keep up the good work.

                        To stay focused on the issue, however, I wonder how challenging it would be to have Spring Integration flag this situation where there is a synchronous pub-sub and when a module throws a "hidden" error then it would write a log warning if there are subsequent components that follow. It would have saved me/us time and no doubt would save others time, unless Gary Russell were around to immediately offer advice....

                        Thanks,
                        @owebeewan

                        Comment


                        • #13
                          Its not necessarily sync vs async but rather the fact that BroadcastingDispatcher was not able to distribute message to all of its subscribers. . . But the flag Gary pointed out is what allows you to manage the severity of this situation. Having said that I just realized that this flag is not mentioned anywhere in the documentation so please feel free to rais a JIRA for that https://jira.springsource.org/browse/INT

                          Comment


                          • #14
                            I was just going through various posts on the forum to have better understanding of pubsub channel and happen to walk through this post as well.
                            I agree with few of the comments that the examples on pubsub channel are not tirivial enough to give a better understanding but let me also tell all the developers those who are using or planning to use Spring Integration f/w that these guys (Mark, Gary, et.al) have been more than helpful in resolving out the issues posted on this forum.
                            I've been member of other so called "poupular" forums as well but never did I experience to get replies to my post (that are meaningful and to-the-point) as I've been getting on this forum and that too "directly" from the team members who are the cofounders of this f/w.
                            I agree totally with Mark that such acusatory tones are doing nothing but defeating the purpose of this forum.

                            I hope all the users of this forum will agree with my comments.

                            Best Regards
                            LB

                            Comment

                            Working...
                            X