Thanks for the reply.
According to the documentation link you provided, the receiver settles messages either by configuring an appropriate incoming-window-size: pn_messenger_set_incoming_window(messenger, N), or by explicitly accepting messages: pn_messenger_accept(messenger, tracker, 0). I believe examples/c/messenger/recv-async is already doing the latter, calling pn_messenger_accept()... https://github.com/apache/qpid-proton/blob/0.12.2/examples/c/messenger/recv-async.c#L80 I tried setting incoming-window-size to 0, rebuilt/retested, but same result - messages not getting dequeued. messenger = pn_messenger(NULL); ++ pn_messenger_set_incoming_window( messenger, 0); pn_messenger_set_blocking(messenger, false); Maybe I'm not understanding something, but after setting the incoming-window size, trace output from recv-async should reflect that change, but I'm not seeing it: [28c28]: -> AMQP [28c28]:0 -> @open(16) [container-id="receiverW0", hostname="0.0.0.0", channel-max=32767] [28c28]:0 -> @begin(17) [next-outgoing-id=0, incoming-window=2147483647, outgoing-window=2147483647] [28c28]:0 -> @attach(18) [name="myqueue", handle=0, role=true, snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) [address="myqueue", durable=0, timeout=0, dynamic=false], target=@target(41) [address="myqueue", durable=0, timeout=0, dynamic=false], initial-delivery-count=0] [28c28]:0 -> @flow(19) [incoming-window=2147483647, next-outgoing-id=0, outgoing-window=2147483647, handle=0, delivery-count=0, link-credit=1024, drain=false] [28c28]: <- AMQP [28c28]:0 <- @open(16) [container-id="localhost", max-frame-size=131072, channel-max=32767, offered-capabilities=@PN_SYMBOL[:"ANONYMOUS-RELAY", :"DELAYED_DELIVERY"], properties={:product="ActiveMQ", :"topic-prefix"="topic://", :"queue-prefix"="queue://", :version="5.16.4", :platform="Java/1.8.0_281"}] [28c28]:0 <- @begin(17) [remote-channel=0, next-outgoing-id=1, incoming-window=16383, outgoing-window=2147483647, handle-max=65535] [28c28]:0 <- @attach(18) [name="myqueue", handle=0, role=false, snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) [address="myqueue"], target=@target(41) [address="myqueue"], incomplete-unsettled=false, initial-delivery-count=0] [28c28]:0 <- @transfer(20) [handle=0, delivery-id=0, delivery-tag=b"\x00", message-format=0, settled=false] (99) "\x00Sp\xd0\x00\x00\x00\x0b\x00\x00\x00\x05BP\x04@BR\x00\x00Ss\xd0\x00\x00\x00>\x00\x00\x00\x0d@@\xa1\x1bamqp://0.0.0.0:5672/myqueue@@@@@\x83\x00\x00\x00\x00\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\xa1\x05HELLO" store.c --------------- I dropped a few DEBUG statements into messenger.c and store.c to see what's happening after pn_messenger_accept() is called in recv-async. I determined that when store.c->pni_store_update() is called, the following condition fails and the method returns without ever handling delivery or updating remote state based on ACCEPTED/REJECTED status. I don't know the significance of this given my limited experience with AMQP and this project, but wanted to mention it nonetheless. if (!pni_store_tracking(store, id)) { return 0; } https://github.com/apache/qpid-proton/blob/0.12.2/proton-c/src/messenger/store.c#L380 Build latest source on Ubuntu ================================ For sanity I switched away from Solaris, built latest from main branch on Ubuntu 18.04/cmake 3.10.2/gcc 7.5. Retested with ActiveMQ-5.16.4, msgr-send with (10) 128 byte messages to 'myq', but to my surprise the broker still does not dequeue messages upon receipt by msgr-recv. msgr-send -a amqp://0.0.0.0:5672/myq -c 10 -b 128 -V msgr-recv -a amqp://0.0.0.0:5672/myq -c 5 -W 0 -e 0 -V Thinking AMQ 5.16.4 might be the issue, I downgraded to ActiveMQ 5.13.4, retested, same issue. Very frustrating. Which test client? =========================== I'm a little confused with the various client programs for sending/receiving simple messages given subtle inconsistencies in their usage. With regard to latest branch - not 0.12.2, which client should I use for manual testing? (these work fine) ~/c/tools/msgr-send ~/c/tools/msgr-recv (this works) ~/c/tools/reactor-send (Can't get this to work: error: acceptor creation failed) ~/c/tools/reactor-recv amqp://0.0.0.0:5672/myq -c 5 Could not get these to work either, tried many combinations of container_id, addr, port, gave up. ~/c/examples/send ~/c/examples/receive myq amqp://0.0.0.0:5672 PN_TRANSPORT_CLOSED: proton:io: Name or service not known - connect to myq:amqp://0.0.0.0:5672 -----Original Message----- From: Robbie Gemmell <robbie.gemm...@gmail.com> Sent: Wednesday, 23 March, 2022 8:08 AM To: users <users@qpid.apache.org> Subject: Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker You have actually narrowed down the probable cause right in your message. The message is not having any actual state applied in the disposition sent, i.e not being 'accepted/released/modified(failed=true)/rejected' etc. To make the broker consider it consumed will need to accept it by sending the accepted state. I would caution you heavily against proceeding with 0.12.2, it is very old and you will be using 'Messenger' API based bits which have long been disregarded, superceded, considered deprecated or even outright deleted for several years no depending on language, starting even before that old release. You likely wont have a good experience with Messenger, there are many reasons it was moved away from. You also probably arent likely to find much help aiding you with it, if only as probably noone remembers anymore anyway. This was the only pointer I could find, https://github.com/apache/qpid-proton/blob/0.12.x/docs/markdown/messenger/message-disposition.md, though I'd note the last section is largely incorrect so who knows how much use the rest of it actually is. On Wed, 23 Mar 2022 at 07:24, <rafblink...@gmail.com> wrote: > > [client] - proton-c: tags\0.12.2 > [broker] - ActiveMQ 5.16.4 / Java SE 1.8.0_281-b09 [HD/OS] - > SPARC-M8 sun4v/Solaris 10 > > Issue > ================================ > Proton-C AMQP 1.0 client consumes messages from ActiveMQ 5.16.4 > broker, but broker does not remove message from the queue and does not > increment 'Dequeue#' count. > > > ================================ > Summary > ================================ > Compiled Proton-C 0.12.2 on Solaris 10/cmake 2.8.0/gcc 4.2 with minor > change to src/posix/io.c [see footnote 1]. Tested simple message > exchange using provided send/recv and send-async/recv-async clients as-is, no > code changes. > Sender produces successfully, no errors. Receiver gets message, prints > body; no errors. But broker never remove message from the queue, nor > does broker increment JMX "Dequeue" stat. My experience with ActiveMQ > and JMS are extensive, but very limited with AMQP protocol, so I don't > know if this issue is caused by not using appropriate delivery and/or > settlement modes - or not accepting or acknowledging messages > correctly in the client, or if the issue resides in the broker. > > Why not build a more recent version of Proton? > Of course we would like to, but current hardware/software constraints > beyond our control prohibit us from using cmake above version 2.8.0 in > Solaris. and the most recent version of the source Proton-C 0.12.2 is > the latest source we can build as later versions of the code require cmake > 3.0+. > > > ================================ > TEST PROCEDURE > ================================ > Test queue: q13 > > [Start Receiver] > ------------------------------------------------ > > recv amqp://0.0.0.0:5672/q13 > [28a98]: -> AMQP > [28a98]:0 -> @open(16) > [container-id="B01C2C42-962E-4C6C-9A05-2841AF174F49", > hostname="0.0.0.0", channel-max=32767] > [28a98]:0 -> @begin(17) [next-outgoing-id=0, > incoming-window=2147483647, outgoing-window=2147483647] > [28a98]:0 -> @attach(18) [name="q13", handle=0, role=true, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13", durable=0, timeout=0, dynamic=false], > target=@target(41) [address="q13", durable=0, timeout=0, > dynamic=false], initial-delivery-count=0] > [28a98]:0 -> @flow(19) [incoming-window=2147483647, > next-outgoing-id=0, outgoing-window=2147483647, handle=0, > delivery-count=0, link-credit=1024, drain=false] > [28a98]: <- AMQP > [28a98]:0 <- @open(16) [container-id="localhost", > max-frame-size=131072, channel-max=32767, > offered-capabilities=@PN_SYMBOL[:"ANONYMOUS-RELAY", > :"DELAYED_DELIVERY"], properties={:product="ActiveMQ", > :"topic-prefix"="topic://", :"queue-prefix"="queue://", > :version="5.16.4", :platform="Java/1.8.0_281"}] > [28a98]:0 <- @begin(17) [remote-channel=0, next-outgoing-id=1, > incoming-window=16383, outgoing-window=2147483647, handle-max=65535] > [28a98]:0 <- @attach(18) [name="q13", handle=0, role=false, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13"], > target=@target(41) [address="q13"], incomplete-unsettled=false, > initial-delivery-count=0] > > > bin/activemq dstat queue (looks ok) > ------------------------------------ > Name Queue Size Producer # Consumer # Enqueue # Dequeue# > q13 0 0 1 > 0 0 > > [Start sender] > --------------------------------------------------------------- > > send -a amqp://0.0.0.0:5672/q13 burninhell > [28ca0]: -> AMQP > [28ca0]:0 -> @open(16) > [container-id="BA01C342-C97D-4C74-A835-0D59F962B608", > hostname="0.0.0.0", channel-max=32767] > [28ca0]:0 -> @begin(17) [next-outgoing-id=0, > incoming-window=2147483647, outgoing-window=2147483647] > [28ca0]:0 -> @attach(18) [name="sender-xxx", handle=0, role=false, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13", durable=0, timeout=0, dynamic=false], > target=@target(41) [address="q13", durable=0, timeout=0, > dynamic=false], initial-delivery-count=0] > [28ca0]: <- AMQP > [28ca0]:0 <- @open(16) [container-id="localhost", > max-frame-size=131072, channel-max=32767, > offered-capabilities=@PN_SYMBOL[:"ANONYMOUS-RELAY", > :"DELAYED_DELIVERY"], properties={:product="ActiveMQ", > :"topic-prefix"="topic://", :"queue-prefix"="queue://", > :version="5.16.4", :platform="Java/1.8.0_281"}] > [28ca0]:0 <- @begin(17) [remote-channel=0, next-outgoing-id=1, > incoming-window=16383, outgoing-window=2147483647, handle-max=65535] > [28ca0]:0 <- @attach(18) [name="sender-xxx", handle=0, role=true, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13"], > target=@target(41) [address="q13"]] > [28ca0]:0 <- @flow(19) [next-incoming-id=0, incoming-window=16383, > next-outgoing-id=1, outgoing-window=2147483647, handle=0, > delivery-count=0, link-credit=1000] > [28ca0]:0 -> @transfer(20) [handle=0, delivery-id=0, > delivery-tag=b"\x00\x00\x00\x00\x00\x00\x00\x00", message-format=0, > settled=true, more=false] (100) > "\x00Sp\xd0\x00\x00\x00\x0b\x00\x00\x00\x05BP\x04@BR\x00\x00Ss\xd0\x00 > \x00\x > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x00\ > x00\x0 > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\xa1 > \x0abu > rninhell" > [28ca0]:0 -> @detach(22) [handle=0, closed=true] > [28ca0]:0 -> @close(24) [] > [28ca0]: -> EOS > [28ca0]:0 <- @detach(22) [handle=0, closed=true] > [28ca0]:0 <- @close(24) [] > [28ca0]: <- EOS > > > [Receiver output - gets message!] > ----------------------------------------- > [28a98]:0 <- @transfer(20) [handle=0, delivery-id=0, > delivery-tag=b"\x00", message-format=0, settled=false] (100) > "\x00Sp\xd0\x00\x00\x00\x0b\x00\x00\x00\x05BP\x04@BR\x00\x00Ss\xd0\x00 > \x00\x > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x00\ > x00\x0 > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\xa1 > \x0abu > rninhell" > Address: amqp://0.0.0.0:5672/q13 > Subject: (no subject) > Content: "burninhell" > [28a98]:0 -> @disposition(21) [role=true, first=0, last=0, > settled=true] > > > > bin/activemq dstat queue > (not looking good, Queue Size should be 0, Dequeue should be 1) > ------------------------------------------ > Name Queue Size Producer # Consumer # Enqueue # Dequeue# > q13 1 0 1 > 1 0 > > > > bin/activemq browse q13 > ------------------------------------------- > JMS_CUSTOM_FIELD:JMS_AMQP_FirstAcquirer = false > JMS_HEADER_FIELD:JMSDestination = q13 JMS_HEADER_FIELD:JMSDeliveryMode > = non-persistent JMS_HEADER_FIELD:JMSMessageID = > ID:gp-das3-44354-1647892116902-3:24:0:0:1 > JMS_CUSTOM_FIELD:JMS_AMQP_NATIVE = true > JMS_CUSTOM_FIELD:JMS_AMQP_HEADER = true > JMS_BODY_FIELD:JMSBytes:1 = > JMS_HEADER_FIELD:JMSExpiration = 0 > JMS_HEADER_FIELD:JMSPriority = 4 > JMS_HEADER_FIELD:JMSRedelivered = false > JMS_CUSTOM_FIELD:JMS_AMQP_PROPERTIES = true > JMS_HEADER_FIELD:JMSTimestamp = > 0 > > > [Terminate receiver] > [Start new receiver] > Picks up same message... > ------------------------------------------------ > > recv amqp://0.0.0.0:5672/q13 > [28a98]: -> AMQP > [28a98]:0 -> @open(16) > [container-id="0A110962-EF33-444C-8509-370551288375", > hostname="0.0.0.0", channel-max=32767] > [28a98]:0 -> @begin(17) [next-outgoing-id=0, > incoming-window=2147483647, outgoing-window=2147483647] > [28a98]:0 -> @attach(18) [name="q13", handle=0, role=true, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13", durable=0, timeout=0, dynamic=false], > target=@target(41) [address="q13", durable=0, timeout=0, > dynamic=false], initial-delivery-count=0] > [28a98]:0 -> @flow(19) [incoming-window=2147483647, > next-outgoing-id=0, outgoing-window=2147483647, handle=0, > delivery-count=0, link-credit=1024, drain=false] > [28a98]: <- AMQP > [28a98]:0 <- @open(16) [container-id="localhost", > max-frame-size=131072, channel-max=32767, > offered-capabilities=@PN_SYMBOL[:"ANONYMOUS-RELAY", > :"DELAYED_DELIVERY"], properties={:product="ActiveMQ", > :"topic-prefix"="topic://", :"queue-prefix"="queue://", > :version="5.16.4", :platform="Java/1.8.0_281"}] > [28a98]:0 <- @begin(17) [remote-channel=0, next-outgoing-id=1, > incoming-window=16383, outgoing-window=2147483647, handle-max=65535] > [28a98]:0 <- @attach(18) [name="q13", handle=0, role=false, > snd-settle-mode=2, rcv-settle-mode=0, source=@source(40) > [address="q13"], > target=@target(41) [address="q13"], incomplete-unsettled=false, > initial-delivery-count=0] > [28a98]:0 <- @transfer(20) [handle=0, delivery-id=0, > delivery-tag=b"\x00", message-format=0, settled=false] (92) > "\x00Sp\xc0\x08\x05BP\x04@BR\x01\x00Ss\xd0\x00\x00\x008\x00\x00\x00\x0 > c@@\xa > 1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x00\x00\x00\x00\x00\x00\x00\ > x83\x0 0\x00\x00\x00\x00\x00\x00\x00@C\x00Sw\xa1\x0aburninhell" > Address: amqp://0.0.0.0:5672/q13 > Subject: (no subject) > Content: "burninhell" > [28a98]:0 -> @disposition(21) [role=true, first=0, last=0, > settled=true] > > > bin/activemq dstat queue > (not looking good, Queue Size should be 0, Dequeue should be 1) > ------------------------------------------ > Name Queue Size Producer # Consumer # Enqueue # Dequeue# > q13 1 0 1 > 1 0 > > bin/activemq browse q13 > (DeliveryCount showing 2, Redelivered=true, so broker interprets > initial consumption of message as a failure) > ------------------------------------------- > JMS_CUSTOM_FIELD:JMS_AMQP_FirstAcquirer = false > JMS_HEADER_FIELD:JMSDestination = q13 JMS_HEADER_FIELD:JMSDeliveryMode > = non-persistent JMS_HEADER_FIELD:JMSMessageID = > ID:gp-das3-44354-1647892116902-3:24:0:0:1 > JMS_CUSTOM_FIELD:JMS_AMQP_NATIVE = true > JMS_CUSTOM_FIELD:JMSXDeliveryCount = 2 <<<<<<<<< > JMS_CUSTOM_FIELD:JMS_AMQP_HEADER = true > JMS_BODY_FIELD:JMSBytes:1 = > JMS_HEADER_FIELD:JMSExpiration = 0 > JMS_HEADER_FIELD:JMSPriority = 4 > JMS_HEADER_FIELD:JMSRedelivered = true <<<<<<<<<< > JMS_CUSTOM_FIELD:JMS_AMQP_PROPERTIES = true > JMS_HEADER_FIELD:JMSTimestamp = 0 > > > =================================== > Observations/Questions > =================================== > Default recv-settle-mode=0, or "AT_MOST_ONCE", this means "fire-and-forget" > from the sender's viewpoint, so it's my understanding there's no > two-way accept/settlement required as in the case of AT_LEAST_ONCE. > Also, each time the receiver consumes and accepts the message the > frame trace shows 'settled=true', so I don't understand why the broker > is not dequeing the message. As I said before, I'm using recv/send and > recv-async/send-async as-is without manipulating message headers or > properties before sending. Do I need to set message headers to > explicitly settle messages and force the broker to dequeue the > message? . I'll admit to being spoiled with JMS and > AUTO_ACKNOWLEDGEMNTs, so perhaps my assumptions were unrealistic to > expect a similar level of ease with AMQP, lol. Any insight will be greatly > appreciated, thanks for reading. > > > ========================================== > Additional info on Build\Make + Footnotes > ========================================== > > Proton-C Client > --------------- > Source: tags\0.12.2 (minor edit to io.c, see footnote 1) > gcc 4.2.0 > cmake 2.8.0 > ld 2.7 (with BFD 2.7) > Solaris 10 > SPARC-M8/sun4v > > [cmake command] > ../cmake-2.8.0-SunOS-sparc/bin/cmake ../qpid-proton-0.11.0 > -DBUILD_PYTHON=OFF -DBUILD_RUBY=OFF -DBUILD_GO=OFF -DBUILD_PHP=OFF > -DBUILD_JAVA=OFF -DBUILD_PERL=OFF -DCMAKE_C_FLAGS="-pthread" > -DCXX_WARNING_FLAGS:STRING="-Wall -pedantic -Wno-format > -Wno-variadic-macros" > -DOPENSSL_SSL_LIBRARIES:FILEPATH=/usr/lib/libssl.so > -DOPENSSL_CRYPTO_LIBRARIES:FILEPATH=/usr/lib/libcrypto.so > -DCYRUS_SASL_LIBRARY:FILEPATH=/usr/local/lib/libsasl2.so > -DCMAKE_SHARED_LINKER_FLAGS="-L/usr/lib -lsocket -lc -lsasl" > -DBUILD_CPP=OFF -DENABLE_WARNING_ERROR:BOOL=OFF > > [make command] > make VERBOSE=1 (duh) > > Broker > ------ > ActiveMQ 5.16.4 > Pre-compiled binaries > Runtime: Java SE 1.8.0_281-b09 > > > Footnotes > ====================================================================== > ====== > ========================== > [1] Solaris does not support MSG_NOSIGNAL or SO_NOSIGPIPE, so cmake > fails in proton-c/src/posix/io.c. > Removed pre-processor directives and implemented minor change in > io.c according to this: > https://markmail.org/message/q436ukyazsfx55av#query:+page:1+mid:vevuwb > h66264 > vcdh+state:results > > > > > > > > --------------------------------------------------------------------- > To unsubscribe, e-mail: users-unsubscr...@qpid.apache.org For > additional commands, e-mail: users-h...@qpid.apache.org > --------------------------------------------------------------------- To unsubscribe, e-mail: users-unsubscr...@qpid.apache.org For additional commands, e-mail: users-h...@qpid.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: users-unsubscr...@qpid.apache.org For additional commands, e-mail: users-h...@qpid.apache.org
