You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@qpid.apache.org by ra...@gmail.com on 2022/03/23 07:24:04 UTC
Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
[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\x0c@@\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:vevuwbh66264
vcdh+state:results
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
RE: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by ra...@gmail.com.
Thanks, Robbie!
We finally got Proton-c with CPP bindings built. We confirmed messages getting dequeued from the broker using simple_send/simple_recv examples. Thanks for the help.
In case anyone's interested, our cmake for building CPP bindings on Solaris 10/gcc 4.2.0/cmake 2.8.0/ld 2.7, it was a b**tch since the last time I built anything in C was 20 years ago in college, lol, good learning experience, nonetheless.
../cmake-2.8.0-SunOS-sparc/bin/cmake ../qpid-proton-0.12.2 -DBUILD_PYTHON=OFF -DBUILD_RUBY=OFF -DBUILD_GO=OFF -DBUILD_PHP=OFF -DBUILD_JAVA=OFF -DBUILD_PERL=OFF -DCMAKE_C_FLAGS="-pthread" -DENABLE_WARNING_ERROR:BOOL=OFF -DOPENSSL_SSL_LIBRARIES:FILEPATH=/usr/lib/libssl.so.1.0.0 -DOPENSSL_CRYPTO_LIBRARIES:FILEPATH=/usr/lib/libcrypto.so.1.0.0 -DCMAKE_SHARED_LINKER_FLAGS="-lsocket -lc -lstdc++"
-----Original Message-----
From: Robbie Gemmell <ro...@gmail.com>
Sent: Monday, 28 March, 2022 12:13 PM
To: users <us...@qpid.apache.org>
Subject: Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Yes, the other languages use the C protocol engine and sometimes IO bits.
https://github.com/apache/qpid-proton/blob/0.37.0/INSTALL.md
On Mon, 28 Mar 2022 at 16:56, <ra...@gmail.com> wrote:
>
> I see, and that makes sense. All this time we were building just C, all bindings were disabled in cmake.
>
> So if we build just the CPP binding, does that require the C source as well?
>
> -----Original Message-----
> From: Robbie Gemmell <ro...@gmail.com>
> Sent: Monday, 28 March, 2022 8:45 AM
> To: users <us...@qpid.apache.org>
> Subject: Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued
> from ActiveMQ broker
>
> 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/tut
> orial_page.html [4]
> https://github.com/apache/qpid-proton/blob/0.37.0/c/examples/send.c#L1
> 69-L173
>
> On Sat, 26 Mar 2022 at 00:19, <ra...@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/messeng
> > er
> > /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/messe
> > ng
> > er/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 <ro...@gmail.com>
> > Sent: Wednesday, 23 March, 2022 8:08 AM
> > To: users <us...@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/mess
> > en ger/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, <ra...@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
> > > \x
> > > 00
> > > \x00\x
> > > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\
> > > x0
> > > 0\
> > > x00\x0
> > > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw
> > > \x
> > > a1
> > > \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
> > > \x
> > > 00
> > > \x00\x
> > > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\
> > > x0
> > > 0\
> > > x00\x0
> > > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw
> > > \x
> > > a1
> > > \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\x0
> > > 0\
> > > x0
> > > c@@\xa
> > > 1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x00\x00\x00\x00\x00\x00\
> > > x0
> > > 0\
> > > 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:ve
> > > vu
> > > wb
> > > h66264
> > > vcdh+state:results
> > >
> > >
> > >
> > >
> > >
> > >
> > >
> > > ------------------------------------------------------------------
> > > --
> > > - To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > > additional commands, e-mail: users-help@qpid.apache.org
> > >
> >
> > --------------------------------------------------------------------
> > - To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
> >
> >
> > --------------------------------------------------------------------
> > - To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> additional commands, e-mail: users-help@qpid.apache.org
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For additional commands, e-mail: users-help@qpid.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by Robbie Gemmell <ro...@gmail.com>.
Yes, the other languages use the C protocol engine and sometimes IO bits.
https://github.com/apache/qpid-proton/blob/0.37.0/INSTALL.md
On Mon, 28 Mar 2022 at 16:56, <ra...@gmail.com> wrote:
>
> I see, and that makes sense. All this time we were building just C, all bindings were disabled in cmake.
>
> So if we build just the CPP binding, does that require the C source as well?
>
> -----Original Message-----
> From: Robbie Gemmell <ro...@gmail.com>
> Sent: Monday, 28 March, 2022 8:45 AM
> To: users <us...@qpid.apache.org>
> Subject: Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
>
> 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, <ra...@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/messeng
> > er/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 <ro...@gmail.com>
> > Sent: Wednesday, 23 March, 2022 8:08 AM
> > To: users <us...@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/messen
> > ger/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, <ra...@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\x
> > > 00
> > > \x00\x
> > > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x0
> > > 0\
> > > x00\x0
> > > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\x
> > > a1
> > > \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\x
> > > 00
> > > \x00\x
> > > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x0
> > > 0\
> > > x00\x0
> > > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\x
> > > a1
> > > \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\x0
> > > 0\
> > > 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:vevu
> > > wb
> > > h66264
> > > vcdh+state:results
> > >
> > >
> > >
> > >
> > >
> > >
> > >
> > > --------------------------------------------------------------------
> > > - To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > > additional commands, e-mail: users-help@qpid.apache.org
> > >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
> >
> >
> > ---------------------------------------------------------------------
> > To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For additional commands, e-mail: users-help@qpid.apache.org
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
> For additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
RE: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by ra...@gmail.com.
I see, and that makes sense. All this time we were building just C, all bindings were disabled in cmake.
So if we build just the CPP binding, does that require the C source as well?
-----Original Message-----
From: Robbie Gemmell <ro...@gmail.com>
Sent: Monday, 28 March, 2022 8:45 AM
To: users <us...@qpid.apache.org>
Subject: Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
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, <ra...@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/messeng
> er/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 <ro...@gmail.com>
> Sent: Wednesday, 23 March, 2022 8:08 AM
> To: users <us...@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/messen
> ger/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, <ra...@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\x
> > 00
> > \x00\x
> > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x0
> > 0\
> > x00\x0
> > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\x
> > a1
> > \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\x
> > 00
> > \x00\x
> > 00:\x00\x00\x00\x0d@@\xa1\x17amqp://0.0.0.0:5672/q13@@@@@\x83\x00\x0
> > 0\
> > x00\x0
> > 0\x00\x00\x00\x00\x83\x00\x00\x00\x00\x00\x00\x00\x00@R\x00@\x00Sw\x
> > a1
> > \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\x0
> > 0\
> > 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:vevu
> > wb
> > h66264
> > vcdh+state:results
> >
> >
> >
> >
> >
> >
> >
> > --------------------------------------------------------------------
> > - To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> additional commands, e-mail: users-help@qpid.apache.org
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For
> additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For additional commands, e-mail: users-help@qpid.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by Robbie Gemmell <ro...@gmail.com>.
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, <ra...@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 <ro...@gmail.com>
> Sent: Wednesday, 23 March, 2022 8:08 AM
> To: users <us...@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, <ra...@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-unsubscribe@qpid.apache.org For
> > additional commands, e-mail: users-help@qpid.apache.org
> >
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For additional commands, e-mail: users-help@qpid.apache.org
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
> For additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
RE: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by ra...@gmail.com.
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 <ro...@gmail.com>
Sent: Wednesday, 23 March, 2022 8:08 AM
To: users <us...@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, <ra...@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-unsubscribe@qpid.apache.org For
> additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org For additional commands, e-mail: users-help@qpid.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org
Re: Messages consumed by Proton-C AMQP 1.0 are not dequeued from ActiveMQ broker
Posted by Robbie Gemmell <ro...@gmail.com>.
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, <ra...@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\x0c@@\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:vevuwbh66264
> vcdh+state:results
>
>
>
>
>
>
>
> ---------------------------------------------------------------------
> To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
> For additional commands, e-mail: users-help@qpid.apache.org
>
---------------------------------------------------------------------
To unsubscribe, e-mail: users-unsubscribe@qpid.apache.org
For additional commands, e-mail: users-help@qpid.apache.org