You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2026/08/06 16:26:10 UTC
(qpid-proton) branch main updated (c1861a920 -> f41881d6b)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
from c1861a920 NO-JIRA: Stop ruby tests creating sasl_conf directory in source tree
new 94e2df384 PROTON-2945: Allow only applicable performatives
new f41881d6b PROTON-2946: Fix interop round trip comparison to be semantic
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
c/src/core/dispatcher.c | 136 ++++++++++++++++++++++++-----------
c/src/core/dispatcher.h | 3 +-
c/src/core/transport.c | 2 +-
c/src/sasl/sasl.c | 3 +-
python/tests/proton_tests/interop.py | 22 +++---
5 files changed, 111 insertions(+), 55 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
(qpid-proton) 01/02: PROTON-2945: Allow only applicable performatives
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit 94e2df384a493cbb8a4870a40d90e08708e131eb
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Tue Aug 4 17:53:35 2026 -0400
PROTON-2945: Allow only applicable performatives
* Only AMQP performatives when dispatching AMQP
* Only SASL performatives when dispatching SASL
---
c/src/core/dispatcher.c | 136 +++++++++++++++++++++++++++++++++---------------
c/src/core/dispatcher.h | 3 +-
c/src/core/transport.c | 2 +-
c/src/sasl/sasl.c | 3 +-
4 files changed, 98 insertions(+), 46 deletions(-)
diff --git a/c/src/core/dispatcher.c b/c/src/core/dispatcher.c
index b9d577c81..4e07889f0 100644
--- a/c/src/core/dispatcher.c
+++ b/c/src/core/dispatcher.c
@@ -39,70 +39,101 @@ int pni_bad_frame_type(pn_transport_t *transport, uint8_t frame_type, uint16_t c
return PN_ERR;
}
+static inline int pni_dispatch_amqp_action(pn_transport_t* transport, uint64_t lcode, uint16_t channel, pn_bytes_t frame_payload)
+{
+ pn_action_t *action;
+ switch (lcode) {
+ case AMQP_DESC_OPEN: action = pn_do_open; break;
+ case AMQP_DESC_BEGIN: action = pn_do_begin; break;
+ case AMQP_DESC_ATTACH: action = pn_do_attach; break;
+ case AMQP_DESC_FLOW: action = pn_do_flow; break;
+ case AMQP_DESC_TRANSFER: action = pn_do_transfer; break;
+ case AMQP_DESC_DISPOSITION: action = pn_do_disposition; break;
+ case AMQP_DESC_DETACH: action = pn_do_detach; break;
+ case AMQP_DESC_END: action = pn_do_end; break;
+ case AMQP_DESC_CLOSE: action = pn_do_close; break;
+ default: action = pni_bad_frame; break;
+ };
+ return action(transport, AMQP_FRAME_TYPE, channel, frame_payload);
+}
+
+static inline int pni_dispatch_sasl_action(pn_transport_t* transport, uint64_t lcode, uint16_t channel, pn_bytes_t frame_payload)
+{
+ pn_action_t *action;
+ switch (lcode) {
+ case AMQP_DESC_SASL_MECHANISMS: action = pn_do_mechanisms; break;
+ case AMQP_DESC_SASL_INIT: action = pn_do_init; break;
+ case AMQP_DESC_SASL_CHALLENGE: action = pn_do_challenge; break;
+ case AMQP_DESC_SASL_RESPONSE: action = pn_do_response; break;
+ case AMQP_DESC_SASL_OUTCOME: action = pn_do_outcome; break;
+ default: action = pni_bad_frame; break;
+ };
+ return action(transport, SASL_FRAME_TYPE, channel, frame_payload);
+}
+
// We could use a table based approach here if we needed to dynamically
// add new performatives
static inline int pni_dispatch_action(pn_transport_t* transport, uint64_t lcode, uint8_t frame_type, uint16_t channel, pn_bytes_t frame_payload)
{
- pn_action_t *action;
switch (frame_type) {
case AMQP_FRAME_TYPE:
- /* Regular AMQP frames */
- switch (lcode) {
- case AMQP_DESC_OPEN: action = pn_do_open; break;
- case AMQP_DESC_BEGIN: action = pn_do_begin; break;
- case AMQP_DESC_ATTACH: action = pn_do_attach; break;
- case AMQP_DESC_FLOW: action = pn_do_flow; break;
- case AMQP_DESC_TRANSFER: action = pn_do_transfer; break;
- case AMQP_DESC_DISPOSITION: action = pn_do_disposition; break;
- case AMQP_DESC_DETACH: action = pn_do_detach; break;
- case AMQP_DESC_END: action = pn_do_end; break;
- case AMQP_DESC_CLOSE: action = pn_do_close; break;
- default: action = pni_bad_frame; break;
- };
- break;
+ return pni_dispatch_amqp_action(transport, lcode, channel, frame_payload);
case SASL_FRAME_TYPE:
- /* SASL frames */
- switch (lcode) {
- case AMQP_DESC_SASL_MECHANISMS: action = pn_do_mechanisms; break;
- case AMQP_DESC_SASL_INIT: action = pn_do_init; break;
- case AMQP_DESC_SASL_CHALLENGE: action = pn_do_challenge; break;
- case AMQP_DESC_SASL_RESPONSE: action = pn_do_response; break;
- case AMQP_DESC_SASL_OUTCOME: action = pn_do_outcome; break;
- default: action = pni_bad_frame; break;
- };
- break;
- default: action = pni_bad_frame_type; break;
+ return pni_dispatch_sasl_action(transport, lcode, channel, frame_payload);
+ default:
+ return pni_bad_frame_type(transport, frame_type, channel, frame_payload);
};
- return action(transport, frame_type, channel, frame_payload);
}
-static int pni_dispatch_frame(pn_frame_t frame, pn_logger_t *logger, pn_transport_t * transport)
+static inline int pni_performative_code(pn_bytes_t frame_payload, uint64_t *lcode)
+{
+ pni_consumer_t consumer = make_consumer_from_bytes(frame_payload);
+ pni_consumer_t subconsumer;
+ if (!consume_described_ulong_descriptor(&consumer, &subconsumer, lcode)
+ || !pni_islist(&subconsumer)
+ ) {
+ return PN_ERR;
+ }
+ return PN_OK;
+}
+
+static inline int pni_dispatch_amqp_frame(pn_frame_t frame, pn_logger_t *logger, pn_transport_t * transport)
{
pn_bytes_t frame_payload = frame.frame_payload0;
if (frame_payload.size == 0) { // ignore null frames
- return 0;
+ return PN_OK;
}
uint64_t lcode;
- pni_consumer_t consumer = make_consumer_from_bytes(frame_payload);
- pni_consumer_t subconsumer;
- if (!consume_described_ulong_descriptor(&consumer, &subconsumer, &lcode)
- || !pni_islist(&subconsumer)
- ) {
+ int err = pni_performative_code(frame_payload, &lcode);
+ if (err != PN_OK) {
PN_LOG(logger, PN_SUBSYSTEM_AMQP, PN_LEVEL_ERROR, "Error dispatching frame");
- return PN_ERR;
+ return err;
}
- uint8_t frame_type = frame.type;
- uint16_t channel = frame.channel;
+ return pni_dispatch_amqp_action(transport, lcode, frame.channel, frame_payload);
+}
+
+static inline int pni_dispatch_sasl_frame(pn_frame_t frame, pn_logger_t *logger, pn_transport_t * transport)
+{
+ pn_bytes_t frame_payload = frame.frame_payload0;
- int err = pni_dispatch_action(transport, lcode, frame_type, channel, frame_payload);
+ if (frame_payload.size == 0) { // ignore null frames
+ return PN_OK;
+ }
+
+ uint64_t lcode;
+ int err = pni_performative_code(frame_payload, &lcode);
+ if (err != PN_OK) {
+ PN_LOG(logger, PN_SUBSYSTEM_AMQP, PN_LEVEL_ERROR, "Error dispatching frame");
+ return err;
+ }
- return err;
+ return pni_dispatch_sasl_action(transport, lcode, frame.channel, frame_payload);
}
-ssize_t pn_dispatcher_input(pn_transport_t *transport, const char *bytes, size_t available, bool batch, bool *halt)
+ssize_t pn_dispatcher_amqp_input(pn_transport_t *transport, const char *bytes, size_t available, bool *halt)
{
size_t read = 0;
@@ -114,7 +145,7 @@ ssize_t pn_dispatcher_input(pn_transport_t *transport, const char *bytes, size_t
read += n;
available -= n;
transport->input_frames_ct += 1;
- int e = pni_dispatch_frame(frame, &transport->logger, transport);
+ int e = pni_dispatch_amqp_frame(frame, &transport->logger, transport);
if (e) return e;
} else if (n < 0) {
pn_do_error(transport, "amqp:connection:framing-error", "malformed frame");
@@ -122,8 +153,29 @@ ssize_t pn_dispatcher_input(pn_transport_t *transport, const char *bytes, size_t
} else {
break;
}
+ }
- if (!batch) break;
+ return read;
+}
+
+ssize_t pn_dispatcher_sasl_input(pn_transport_t *transport, const char *bytes, size_t available, bool *halt)
+{
+ size_t read = 0;
+
+ if (available && !*halt) {
+ pn_frame_t frame;
+
+ ssize_t n = pn_read_frame(&frame, bytes + read, available, transport->local_max_frame, &transport->logger);
+ if (n > 0) {
+ read += n;
+ available -= n;
+ transport->input_frames_ct += 1;
+ int e = pni_dispatch_sasl_frame(frame, &transport->logger, transport);
+ if (e) return e;
+ } else if (n < 0) {
+ pn_do_error(transport, "amqp:connection:framing-error", "malformed frame");
+ return n;
+ }
}
return read;
diff --git a/c/src/core/dispatcher.h b/c/src/core/dispatcher.h
index 2d3cf2134..46466aba6 100644
--- a/c/src/core/dispatcher.h
+++ b/c/src/core/dispatcher.h
@@ -31,7 +31,8 @@
typedef int (pn_action_t)(pn_transport_t *transport, uint8_t frame_type, uint16_t channel, const pn_bytes_t frame_payload);
-ssize_t pn_dispatcher_input(pn_transport_t* transport, const char* bytes, size_t available, bool batch, bool* halt);
+ssize_t pn_dispatcher_amqp_input(pn_transport_t* transport, const char* bytes, size_t available, bool* halt);
+ssize_t pn_dispatcher_sasl_input(pn_transport_t* transport, const char* bytes, size_t available, bool* halt);
ssize_t pn_dispatcher_output(pn_transport_t *transport, char *bytes, size_t size);
#endif /* dispatcher.h */
diff --git a/c/src/core/transport.c b/c/src/core/transport.c
index 313e96ada..e074c6682 100644
--- a/c/src/core/transport.c
+++ b/c/src/core/transport.c
@@ -2619,7 +2619,7 @@ static ssize_t pn_input_read_amqp(pn_transport_t* transport, unsigned int layer,
}
- ssize_t n = pn_dispatcher_input(transport, bytes, available, true, &transport->halt);
+ ssize_t n = pn_dispatcher_amqp_input(transport, bytes, available, &transport->halt);
if (n < 0 || transport->close_rcvd) {
return PN_EOS;
} else {
diff --git a/c/src/sasl/sasl.c b/c/src/sasl/sasl.c
index 7f0bc21e2..3e639de58 100644
--- a/c/src/sasl/sasl.c
+++ b/c/src/sasl/sasl.c
@@ -636,7 +636,7 @@ static ssize_t pn_input_read_sasl(pn_transport_t* transport, unsigned int layer,
pni_sasl_start_server_if_needed(transport);
if (!pni_sasl_is_final_input_state(sasl)) {
- ssize_t n = pn_dispatcher_input(transport, bytes, available, false, &transport->halt);
+ ssize_t n = pn_dispatcher_sasl_input(transport, bytes, available, &transport->halt);
if (n < 0 || transport->close_rcvd) {
return PN_EOS;
} else {
@@ -1065,4 +1065,3 @@ int pn_do_outcome(pn_transport_t *transport, uint8_t frame_type, uint16_t channe
pnx_sasl_set_desired_state(transport, authenticated ? SASL_RECVED_SUCCESS : SASL_RECVED_FAILURE);
return 0;
}
-
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
(qpid-proton) 02/02: PROTON-2946: Fix interop round trip comparison to be semantic
Posted by as...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit f41881d6be1b59f7de370b1a21d6a9885a0cc22b
Author: Andrew Stitcher <as...@apache.org>
AuthorDate: Wed Jul 22 17:35:04 2026 -0400
PROTON-2946: Fix interop round trip comparison to be semantic
Previously we used a byte for byte comparison but it is legitimate to encode using small encodings where the value will fit even if the bytes decoded used a large encoding.
---
python/tests/proton_tests/interop.py | 22 +++++++++++++---------
1 file changed, 13 insertions(+), 9 deletions(-)
diff --git a/python/tests/proton_tests/interop.py b/python/tests/proton_tests/interop.py
index 97eb17e5a..2271471ea 100644
--- a/python/tests/proton_tests/interop.py
+++ b/python/tests/proton_tests/interop.py
@@ -54,29 +54,33 @@ class InteropTest(common.Test):
finally:
f.close()
- def decode_data(self, encoded):
+ def decode_data(self, encoded, decoded):
buffer = encoded
while buffer:
- n = self.data.decode(buffer)
+ n = decoded.decode(buffer)
buffer = buffer[n:]
- self.data.rewind()
+ decoded.rewind()
def decode_data_file(self, name):
encoded = self.get_data(name)
- self.decode_data(encoded)
+ self.decode_data(encoded, self.data)
encoded_size = self.data.encoded_size()
# Re-encode and verify pre-computed and actual encoded size match.
reencoded = self.data.encode()
assert encoded_size == len(reencoded), "%d != %d" % (encoded_size, len(reencoded))
- # verify round trip bytes
- assert reencoded == encoded, "Value mismatch: %s != %s" % (reencoded, encoded)
+ # Verify semantic round-trip: decode the re-encoded bytes and compare
+ # formatted output. Byte-for-byte equality is not required because the
+ # encoder may legitimately choose a more compact encoding (e.g. LIST8
+ # instead of LIST32) that is semantically identical.
+ reencoded_data = Data()
+ self.decode_data(reencoded, reencoded_data)
+ assert self.data.format() == reencoded_data.format(), \
+ "Value mismatch: %s != %s" % (self.data.format(), reencoded_data.format())
def decode_message_file(self, name):
self.message.decode(self.get_data(name))
body = self.message.body
- if str(type(body)) == "<type 'org.apache.qpid.proton.amqp.Binary'>":
- body = body.array.tostring()
- self.decode_data(body)
+ self.decode_data(body, self.data)
def assert_next(self, type, value):
next_type = self.data.next()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org