🎄 Happy Holidays! 🥳

Most of Solace is closed December 24–January 1 so our employees can spend time with their families. We will re-open Thursday, January 2, 2024. Please expect slower response times during this period and open a support ticket for anything needing immediate assistance.

Happy Holidays!

Please note: most of Solace is closed December 25–January 2, and will re-open Tuesday, January 3, 2023.

Consuming message from specific Topic subscription added in Queue

Amol D
Amol D Member Posts: 7 ✭
in PubSub+ Event Broker #1

Hi,

I have created the queue with topic subscriptions(more than 1 topic subscription), as below.

"queueName": "QUE_TEST"

 "subscription_topics": [

    "Demo/Msg",

    "Demo"

   ]

now, I am interested in consuming message published to specific topic, like from 'Demo/Msg'. I have created the flow with queue to consume the messages published, so its giving me all messages published on topics under queue, no idea that which message is from which topic. Below is code I have implemented, how to get messages from specific topic under queue?

           var endpoint = ContextFactory.Instance.CreateQueue(queue);      

          var Flow = InternalSolaceSession.CreateFlow(new FlowProperties()

          {

            AckMode = MessageAckMode.AutoAck

          }, endpoint, null, HandleMessageEvent, HandleFlowEvent);

          Flow.Start();

Comments

  • arih
    arih Member, Employee Posts: 125 Solace Employee
    #2

    Hi @Amol D


    You can still find out the topic that particular message was published. I'm guessing for dotnet you can use the IMessage.Destination properties (ref)

    But...

    I had to ask, why would you want to do that? What is the objective?

    A couple of points came to mind:

    • The idea of having Queues as your 'consumption' endpoint with topic subscriptions - means you get the flexibility of adding/changing topic subscriptions without any impact on the consumer apps.
    • If you'd like to subscribe directly to such topics, but do not require persistency, then just do a direct topic subscription. Here's the tutorial for dotnet for reference.
    • Or, you could just create another Queue for that specific topic subscription :)


    -Ari

  • Amol D
    Amol D Member Posts: 7 ✭
    #3

    Thanks @arih

    Since we wanted message persistency and multiple consumer support, we have opted queues with topic subscription.

    I am able to get the topic name with the help of IMessage.Destination properties. however I was keen to understand how I can apply filtering while subscribing the queue.

  • arih
    arih Member, Employee Posts: 125 Solace Employee
    #4

    Hi @Amol D ,


    I recommend to think about multiple queues for different consumers - in other words make more fine-grained Queues :)

    We can get the original destination topic, but I don't recommend to use that to 'filter'. The filtering should happen when the routing is happening within the broker / event mesh.

  • Amol D
    Amol D Member Posts: 7 ✭
    #5

    Thanks @arih , I will think about multiple queue.

    one more question(using .net), I using client ack mode to send the message acknowledgment, however, however its not working as messages are still there on solace, below is the consume and message handler code.

     Would you advise here?

    image.png


    image.png


  • arih
    arih Member, Employee Posts: 125 Solace Employee
    #6

    Not an expert on dotnet, but don't you need to use WaitEventWaitHandle.set() method after the Ack?

  • nicholasdgoodman
    nicholasdgoodman Member, Employee Posts: 43 Solace Employee
    #7

    @Amol D

    In the source code you have shown, It is unclear how you are persisting the flow instance. In SubscribeAndHandle() I see you have declared a local variable, var Flow, and in the HandleMessageEvent callback, there is a class field _flow .

    Is there some part of the code which is missing here?

    Otherwise, the code is generally correct; however, with some caveats that the local queue variable on which the flow is based may be disposed prematurely as soon as the SubscribeAndHandle() method completes. I am not sure what effect that might have on the flow instance.

    Can you verify that the _flow.Ack(...) call is returning ReturnCode.SOLCLIENT_OK?

Categories

This Month's Leaders

This Week's Leaders