Announcement Announcement Module
Collapse
No announcement yet.
How to poll messages from an inbound message adapter Page Title Module
Move Remove Collapse
X
Conversation Detail Module
Collapse
  • Filter
  • Time
  • Show
Clear All
new posts

  • How to poll messages from an inbound message adapter

    We have spring integration code running on multiple jbosses. Each jboss is listening to an inbound hornet Q and the integration flow is like this:

    JMS Inbound adapter -> default channel -> service activator

    We don't want service activator to be processing two different messages at the same time in 2 different JVMs. We plan to implement a distributed lock using hazelcast. The lock will give exclusive access to only 1 jvm at a time to process incoming message.

    My first thought is to implement a custom poller for the inbound adapter, where the custom code can acquire the lock first and then read the inbound message and pass it on for processing. If this is the correct approach, then how do I implement a custom poller for the 'message-driven-channel-adapter'

    Thanks

  • #2
    You can add an advice-chain to the poller (via the <poller> element's advice-chain attribute), and that allows you to provide Advice that wraps the poll operation. There you could add the lock acquire(before)/release(after) behavior.

    There might be other options, but that is the first thing that comes to mind.

    Can you explain why you want to serialize the message processing?

    Comment


    • #3
      Thank you so much for a quick response!

      Originally posted by Mark Fisher View Post
      Can you explain why you want to serialize the message processing?
      The incoming messages from the hornet Q are going to be large xml content, that cause thousands of inserts/deletes in DB. So if multiple such messages arrive around the same time then it's possible the processing of the first one hasn't completed and the inserts/deletes start happening for the second one in parallel. If not serialized we get Oracle errors like "resource busy and acquire with nowait specified".

      It's not critical for us to consume such messages in parallel and serializing is acceptable for low stress on DB.

      Comment


      • #4
        If you are using 2.2.x and, if you need access to the message content (so perhaps you can do conditional locking - say, based on the size of the message), instead of advising the poller, you can add an advice the <service-activator/> instead (request-handler-advice-chain).

        http://static.springsource.org/sprin...r-advice-chain

        http://static.springsource.org/sprin...#custom-advice

        Comment


        • #5
          It does sound like you might want to consider only having a single active poller (by setting auto-startup=false but manually starting one of them, maybe via a control bus message). Then you would need to add the behavior to start another adapter instance if the process hosting the active one fails, but that can be done.

          Comment


          • #6
            Originally posted by Gary Russell View Post
            If you are using 2.2.x and, if you need access to the message content (so perhaps you can do conditional locking - say, based on the size of the message), instead of advising the poller, you can add an advice the <service-activator/> instead (request-handler-advice-chain).
            This is also a good idea and we thought about this, but the issue is that if one JVM is processing a large xml, while the other is being waited on in a different jvm (via a distributed lock), then we might lose the waiting message if jboss crashes or restarts.

            We can implement this idea if there is a guarantee or a mechanism, so that the waited message, until fully processed, will stay in the source hornet queue in case the jvm crashes.

            Comment


            • #7
              No; you should be good - as long as you use the default (auto) ack mode (or transactions), because the container won't ack the message until after it is processed.

              So, if the JVM with the waiting thread crashes, the broker will redeliver the message to another client.

              Comment


              • #8
                Good evening, all!
                Let me to intervene in the debate
                I'm not well with hazelcast, but I see this one: http://www.hazelcast.com/docs/2.5/ma...html/ch09.html
                and here is a sample how to configure <hz:executor-service>: http://www.onlinetechvision.com/?p=644
                From other side Spring Integration's <poller> has an attribute task-executor.
                So you can configure:
                1. single-thread hz:executor-service
                2. and poller:
                HTML Code:
                <jms:inbound-channel-adapter>
                	<si:poller fixed-delay="1000" task-executor="hzExecutorService">
                		<si:transactional/>
                	</si:poller>
                </jms:inbound-channel-adapter>
                It looks like only one message will be processed at the same time per cluster.
                I haven't tested it, but in theory it should work.

                Take care,
                Artem

                Comment

                Working...
                X