How to setup the ack timer for guarantee messaging

https://imgur.com/2SsWWAc

Hi @TomF, thanks for your replies. Please allow me to describe my question again with code.

When received the message, I set the thread.sleep before ack, because I want to simulate the consumer already did some tasks (e.g. parse data, insert into database and so on), means the consumer will not ack to broker immediately.
https://imgur.com/ziZzWmM

private void HandleMessageEvent(object source, MessageEventArgs args)

    {
        using (IMessage message = args.Message)
        {
            // Expecting the message content as a binary attachment
            ...
            //set ack delay
            Thread.Sleep(Convert.ToInt32(this.txtAckdelay.Text));
            // ACK the message
            Flow.Ack(message.ADMessageId);

Based on the sample code of producer, it sends different data by loop, so the consumer will late to ack for each data.

for (int i = 0; i < TotalMessages; i++)

            {
                // Create the message content as a binary attachment
                message.BinaryAttachment = Encoding.ASCII.GetBytes(
                    string.Format("Topic to Queue Mapping Tutorial! Message ID: #{0} data{1}", idx, i));

                // Send the message to the queue on the Solace messaging router
                SetText("Sending message ID " + i + " to topic " + tutorialTopic.Name + " mapped to queue " + queueName + "...");

                ReturnCode returnCode = session.Send(message);

Then I set the ack delay time as 7 seconds, you can see from the following image, the queue data could not be deleted and the consumer received data continuously. That why I would like to adjust the ack timer, because I was guessing that the broker thought the ack was late, then consumer may not successfully received the data.

https://imgur.com/oWEdgAk

Do you have any suggestion if I don't setup the session properties? (In fact, I tried many session properties, did not resolve the situation that I met). Basically, my code is refer to the sample code from Solace, just separated to Producer and Consumer, and for load balancing purpose, I set access type as NonExclusive.

AccessType = EndpointProperties.EndpointAccessType.NonExclusive

Hi @TomF, yes, the Ackmode is ClinetAck from the beginning

Flow = session.CreateFlow(new FlowProperties()

            {
                AckMode = MessageAckMode.ClientAck
            },
            queue, null, HandleMessageEvent, HandleFlowEvent);
            Flow.Start();

And I am sure that the queue data did not be deleted once the consumer late to ack, so the broker sent data again and again.
Thanks for the suggestion, I will see how to implement it to the code.

Hi @TomF, my code as below, since I am just trying to figure out how Solace work with Topic to Queue Mapping, so basically I follow the sample code (https://github.com/SolaceSamples/solace-samples-dotnet/blob/master/src/TopicToQueueMapping/TopicToQueueMapping.cs) to modify it to window form without big change,
and separated to producer and consumer. I even marked cleanup, so should not destroy the flow.

//finally

        //{
        //    // Dispose Solace Systems Messaging API
        //    ContextFactory.Instance.Cleanup();
        //}
        //SetText("Finished.");

//Code started from click button

private void btnSub_Click(object sender, EventArgs e)

    {
        Thread t1 = new Thread(Run);
        t1.Start();
    }

// Initialize Solace Systems Messaging API with logging to console at Warning level

      ContextFactoryProperties cfp = new ContextFactoryProperties()

        {
            SolClientLogLevel = SolLogLevel.Debug
        };
        cfp.LogToConsoleError();
        ContextFactory.Instance.Init(cfp);

        try
        {
            // Context must be created first
            using (IContext context = ContextFactory.Instance.CreateContext(new ContextProperties(), null))
            {
            ...
            // Create session properties
                SessionProperties sessionProps = new SessionProperties()
                {
                    Host = host,
                    VPNName = vpnname,
                    UserName = username,
                    Password = password,
                    ReconnectRetries = DefaultReconnectRetries,
                    IgnoreDuplicateSubscriptionError = true
                };

                sessionProps.GdWithWebTransport = true;
                sessionProps.AckEventMode = SessionProperties.AckMode.PER_MSG;
                sessionProps.ADPublishAckTimerInMsecs = 60000;

                // Connect to the Solace messaging router
                Console.WriteLine("Connecting as {0}@{1} on {2}...", username, vpnname, host);
                using (ISession session = context.CreateSession(sessionProps, null, null))
                {
                    returnCode = session.Connect();
                    if (returnCode == ReturnCode.SOLCLIENT_OK)
                    {
                        Console.WriteLine("Session successfully connected.");

                        if (session.IsCapable(CapabilityType.PUB_GUARANTEED) &&
                            session.IsCapable(CapabilityType.SUB_FLOW_GUARANTEED) &&
                            session.IsCapable(CapabilityType.ENDPOINT_MANAGEMENT) &&
                            session.IsCapable(CapabilityType.QUEUE_SUBSCRIPTIONS))
                        {
                            SetText("All required capabilities supported.");
                            TestConsumer(session);

private void TestConsumer(ISession session)

    {
        // Provision the queue
        string queueName = "testqueue3";

        // Create the queue
        using (IQueue queue = ContextFactory.Instance.CreateQueue(queueName))
        {
            // Set queue permissions to "consume" and access-type to "exclusive"
            EndpointProperties endpointProps = new EndpointProperties()
            {
                Permission = EndpointProperties.EndpointPermission.Consume,
                AccessType = EndpointProperties.EndpointAccessType.NonExclusive,
            };
            // Provision it, and do not fail if it already exists
            session.Provision(queue, endpointProps,
                ProvisionFlag.IgnoreErrorIfEndpointAlreadyExists | ProvisionFlag.WaitForConfirm, null);
            SetText("Queue " + queueName + " has been created and provisioned.");

            // Add subscription to the topic mapped to the queue
            ITopic tutorialTopic = ContextFactory.Instance.CreateTopic("T/mapped/topic/sampleMsg");
            session.Subscribe(queue, tutorialTopic, SubscribeFlag.WaitForConfirm, null);

            Flow = session.CreateFlow(new FlowProperties()
            {
                AckMode = MessageAckMode.ClientAck
            },
            queue, null, HandleMessageEvent, HandleFlowEvent);
            Flow.Start();

            // block the current thread until a confirmation received
            CountdownEvent.Wait();
        }
    }

private void HandleMessageEvent(object source, MessageEventArgs args)

    {
        // Received a message
        Console.WriteLine("Received message.");
        using (IMessage message = args.Message)
        {
            SetText(Encoding.ASCII.GetString(message.BinaryAttachment));
            // ACK the message
            Thread.Sleep(Convert.ToInt32(this.txtAckdelay.Text));
            //args.Message.AckImmediately = true;
            Flow.Ack(message.ADMessageId);
            //CountdownEvent.Signal();
        }
    }

public void HandleFlowEvent(object sender, FlowEventArgs args)

    {
        // Received a flow event
        SetText("Received Flow Event "+ args.Event + " Type: " + args.ResponseCode.ToString() + " Text: " + args.Info);
    }       

Hi @TomF, thanks for the replies.
The purpose of SetText is just to show the information on the window form. I did not overwrite the message, because based on the sample code, it uses the following code to display the message, I just use callback to let it show on the UI.
[Sample code]

Console.WriteLine("Message content: {0}", Encoding.ASCII.GetString(message.BinaryAttachment));

[My code]

SetText(Encoding.ASCII.GetString(message.BinaryAttachment));

delegate void SetTextCallback(string text);

private void SetText(string text)

    {
        if (this.txtSubData.InvokeRequired)
        {
            SetTextCallback d = new SetTextCallback(SetText);
            this.Invoke(d, new object[] { text });
        }
        else
        {
            string time = DateTime.Now.ToString();
            this.txtSubData.Text += time + " " + text + "\r\n";
        }
    }