I dont believe it says you must do either 'pn_messenger_set_incoming_window' or 'pn_messenger_accept', it says you should do both. By not setting a window the message presumably isnt actually tracked and so the code you pointed at just returns as it cant do anything, and you dont actually set an accepted state.
In any case, as I said you should probably avoid the old Messenger API. It has various issues such as this (and particularly around error handling), anything left of it is long deprecated, and noone has worked on it for years. As to which test prog, the examples are going to be the intended examples (note they dont use Messenger), e.g send.c [1]. The main focus in C I believe is really the lower level protocol-engine (what proton started as) and IO bits though, rather than higher level 'client' API which are mainly covered in other languages, e.g see the C++ binding [2][3]. Regardless, you look to be feeding the C example the wrong format of args as e.g you are supplying a URL below when that particular set of example (unlike some of the others) are looking for separate individual host and port arg values, e.g from within send.c [4]: app.container_id = argv[0]; /* Should be unique */ app.host = (argc > 1) ? argv[1] : ""; app.port = (argc > 2) ? argv[2] : "amqp"; app.amqp_address = (argc > 3) ? argv[3] : "examples"; app.message_count = (argc > 4) ? atoi(argv[4]) : 10; [1] https://github.com/apache/qpid-proton/blob/0.37.0/c/examples/send.c [2] https://github.com/apache/qpid-proton/blob/0.37.0/cpp/examples/simple_send.cpp [3] https://qpid.apache.org/releases/qpid-proton-0.37.0/proton/cpp/api/tutorial_page.html [4] https://github.com/apache/qpid-proton/blob/0.37.0/c/examples/send.c#L169-L173 On Sat, 26 Mar 2022 at 00:19, <rafblink...@gmail.com> wrote: > > > 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 > --------------------------------------------------------------------- To unsubscribe, e-mail: users-unsubscr...@qpid.apache.org For additional commands, e-mail: users-h...@qpid.apache.org
