This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
commit 99923055e8cd991b74a58b35c195e11a5051497a
Author: Andrew Stitcher <astitc...@apache.org>
AuthorDate: Wed Aug 7 11:27:52 2024 -0400
PROTON-2873: Reorganize dispositions representation
The natural representation for dispositions is a union because different
types of disposition can only have quite different pieces of data.
---
c/src/core/emitters.h | 103 ++++++++++++--------
c/src/core/engine-internal.h | 41 ++++++--
c/src/core/engine.c | 186 ++++++++++++++++++++++++++++--------
c/src/core/transport.c | 126 +++++++++++++-----------
python/proton/_delivery.py | 9 +-
python/tests/proton_tests/engine.py | 28 ------
6 files changed, 317 insertions(+), 176 deletions(-)
diff --git a/c/src/core/emitters.h b/c/src/core/emitters.h
index f92425b62..2cd59a64c 100644
--- a/c/src/core/emitters.h
+++ b/c/src/core/emitters.h
@@ -614,58 +614,81 @@ static inline void emit_condition(pni_emitter_t* emitter,
pni_compound_context*
}
}
+static inline void emit_received_disposition(pni_emitter_t* emitter,
pni_compound_context* compound0, pn_received_disposition_t *disposition) {
+ for (bool small_encoding = true; ; small_encoding = false) {
+ pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
+ pni_compound_context compound = c;
+ emit_uint(emitter, &compound, disposition->section_number);
+ emit_ulong(emitter, &compound, disposition->section_offset);
+ emit_end_list(emitter, &compound, small_encoding);
+ if (encode_succeeded(emitter, &compound)) break;
+ }
+}
+
+static inline void emit_rejected_disposition(pni_emitter_t* emitter,
pni_compound_context* compound0, pn_rejected_disposition_t *disposition) {
+ for (bool small_encoding = true; ; small_encoding = false) {
+ pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
+ pni_compound_context compound = c;
+ emit_condition(emitter, &compound, &disposition->condition);
+ emit_end_list(emitter, &compound, small_encoding);
+ if (encode_succeeded(emitter, &compound)) break;
+ }
+}
+
+static inline void emit_modified_disposition(pni_emitter_t* emitter,
pni_compound_context* compound0, pn_modified_disposition_t *disposition){
+ for (bool small_encoding = true; ; small_encoding = false) {
+ pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
+ pni_compound_context compound = c;
+ emit_bool(emitter, &compound, disposition->failed);
+ emit_bool(emitter, &compound, disposition->undeliverable);
+ emit_copy_or_raw(emitter, &compound, disposition->annotations,
disposition->annotations_raw);
+ emit_end_list(emitter, &compound, small_encoding);
+ if (encode_succeeded(emitter, &compound)) break;
+ }
+}
+
+static inline void emit_custom_disposition(pni_emitter_t* emitter,
pni_compound_context* compound0, pn_custom_disposition_t *disposition){
+ emit_descriptor(emitter, compound0, disposition->type);
+ if ((disposition->data && pn_data_size(disposition->data) == 0) ||
+ (!disposition->data && disposition->data_raw.size == 0)) {
+ emit_list0(emitter, compound0);
+ return;
+ }
+ pni_compound_context c = make_compound();
+ emit_copy_or_raw(emitter, &c, disposition->data, disposition->data_raw);
+ compound0->count++;
+}
+
static inline void emit_disposition(pni_emitter_t* emitter,
pni_compound_context* compound0, pn_disposition_t *disposition)
{
- if (!disposition || !disposition->type) {
+ if (!disposition || pn_disposition_type(disposition)==PN_DISP_EMPTY) {
emit_null(emitter, compound0);
return;
}
- emit_descriptor(emitter, compound0, disposition->type);
switch (disposition->type) {
- case PN_RECEIVED:
- for (bool small_encoding = true; ; small_encoding = false) {
- pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
- pni_compound_context compound = c;
- emit_uint(emitter, &compound, disposition->section_number);
- emit_ulong(emitter, &compound, disposition->section_offset);
- emit_end_list(emitter, &compound, small_encoding);
- if (encode_succeeded(emitter, &compound)) break;
- }
+ case PN_DISP_RECEIVED:
+ emit_descriptor(emitter, compound0, AMQP_DESC_RECEIVED);
+ emit_received_disposition(emitter, compound0,
&disposition->u.s_received);
+ return;
+ case PN_DISP_ACCEPTED:
+ emit_descriptor(emitter, compound0, AMQP_DESC_ACCEPTED);
+ emit_list0(emitter, compound0);
return;
- case PN_ACCEPTED:
- case PN_RELEASED:
+ case PN_DISP_RELEASED:
+ emit_descriptor(emitter, compound0, AMQP_DESC_RELEASED);
emit_list0(emitter, compound0);
return;
- case PN_REJECTED:
- for (bool small_encoding = true; ; small_encoding = false) {
- pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
- pni_compound_context compound = c;
- emit_condition(emitter, &compound, &disposition->condition);
- emit_end_list(emitter, &compound, small_encoding);
- if (encode_succeeded(emitter, &compound)) break;
- }
+ case PN_DISP_REJECTED:
+ emit_descriptor(emitter, compound0, AMQP_DESC_REJECTED);
+ emit_rejected_disposition(emitter, compound0,
&disposition->u.s_rejected);
return;
- case PN_MODIFIED:
- for (bool small_encoding = true; ; small_encoding = false) {
- pni_compound_context c = emit_list(emitter, compound0, small_encoding,
true);
- pni_compound_context compound = c;
- emit_bool(emitter, &compound, disposition->failed);
- emit_bool(emitter, &compound, disposition->undeliverable);
- emit_copy_or_raw(emitter, &compound, disposition->annotations,
disposition->annotations_raw);
- emit_end_list(emitter, &compound, small_encoding);
- if (encode_succeeded(emitter, &compound)) break;
- }
+ case PN_DISP_MODIFIED:
+ emit_descriptor(emitter, compound0, AMQP_DESC_MODIFIED);
+ emit_modified_disposition(emitter, compound0,
&disposition->u.s_modified);
return;
- default:
- if ((disposition->data && pn_data_size(disposition->data) == 0) ||
- (!disposition->data && disposition->data_raw.size == 0)) {
- emit_list0(emitter, compound0);
- return;
- }
- pni_compound_context c = make_compound();
- emit_copy_or_raw(emitter, &c, disposition->data, disposition->data_raw);
- compound0->count++;
+ case PN_DISP_CUSTOM:
+ emit_custom_disposition(emitter, compound0, &disposition->u.s_custom);
return;
}
}
diff --git a/c/src/core/engine-internal.h b/c/src/core/engine-internal.h
index 62aa12434..e0e588b58 100644
--- a/c/src/core/engine-internal.h
+++ b/c/src/core/engine-internal.h
@@ -333,17 +333,46 @@ struct pn_link_t {
bool more_pending;
};
-struct pn_disposition_t {
+typedef enum pn_disposition_type_t {
+ PN_DISP_EMPTY = 0,
+ PN_DISP_CUSTOM = 1,
+ PN_DISP_RECEIVED = PN_RECEIVED,
+ PN_DISP_ACCEPTED = PN_ACCEPTED,
+ PN_DISP_REJECTED = PN_REJECTED,
+ PN_DISP_RELEASED = PN_RELEASED,
+ PN_DISP_MODIFIED = PN_MODIFIED,
+} pn_disposition_type_t;
+
+typedef struct pn_received_disposition_t {
+ uint64_t section_offset;
+ uint32_t section_number;
+} pn_received_disposition_t;
+
+typedef struct pn_rejected_disposition_t {
pn_condition_t condition;
- uint64_t type;
- pn_data_t *data;
- pn_bytes_t data_raw;
+} pn_rejected_disposition_t;
+
+typedef struct pn_modified_disposition_t {
pn_data_t *annotations;
pn_bytes_t annotations_raw;
- uint64_t section_offset;
- uint32_t section_number;
bool failed;
bool undeliverable;
+} pn_modified_disposition_t;
+
+typedef struct pn_custom_disposition_t {
+ pn_data_t *data;
+ uint64_t type;
+ pn_bytes_t data_raw;
+} pn_custom_disposition_t;
+
+struct pn_disposition_t {
+ union {
+ struct pn_received_disposition_t s_received;
+ struct pn_rejected_disposition_t s_rejected;
+ struct pn_modified_disposition_t s_modified;
+ struct pn_custom_disposition_t s_custom;
+ } u;
+ uint16_t type;
bool settled;
};
diff --git a/c/src/core/engine.c b/c/src/core/engine.c
index 4ad8c4b05..ad9e77fd5 100644
--- a/c/src/core/engine.c
+++ b/c/src/core/engine.c
@@ -26,6 +26,7 @@
#include "consumers.h"
#include "core/frame_consumers.h"
+#include "emitters.h"
#include "fixed_string.h"
#include "framing.h"
#include "memory.h"
@@ -1577,11 +1578,26 @@ pn_session_t *pn_link_session(pn_link_t *link)
static void pn_disposition_finalize(pn_disposition_t *ds)
{
- pn_free(ds->data);
- pn_bytes_free(ds->data_raw);
- pn_free(ds->annotations);
- pn_bytes_free(ds->annotations_raw);
- pn_condition_tini(&ds->condition);
+ switch (ds->type) {
+ case PN_DISP_EMPTY:
+ break;
+ case PN_DISP_RECEIVED:
+ break;
+ case PN_DISP_MODIFIED:
+ pn_data_free(ds->u.s_modified.annotations);
+ pn_bytes_free(ds->u.s_modified.annotations_raw);
+ break;
+ case PN_DISP_REJECTED:
+ pn_condition_tini(&ds->u.s_rejected.condition);
+ break;
+ case PN_DISP_ACCEPTED:
+ case PN_DISP_RELEASED:
+ break;
+ case PN_DISP_CUSTOM:
+ pn_data_free(ds->u.s_custom.data);
+ pn_bytes_free(ds->u.s_custom.data_raw);
+ break;
+ }
}
static void pn_delivery_incref(void *object)
@@ -1656,28 +1672,13 @@ static void pn_delivery_finalize(void *object)
static void pn_disposition_init(pn_disposition_t *ds)
{
- ds->data = NULL;
- ds->data_raw = (pn_bytes_t){0, NULL};
- ds->annotations = NULL;
- ds->annotations_raw = (pn_bytes_t){0, NULL};
- pn_condition_init(&ds->condition);
+ memset(ds, 0, sizeof(*ds));
}
static void pn_disposition_clear(pn_disposition_t *ds)
{
- ds->type = 0;
- ds->section_number = 0;
- ds->section_offset = 0;
- ds->failed = false;
- ds->undeliverable = false;
- ds->settled = false;
- pn_data_clear(ds->data);
- pn_bytes_free(ds->data_raw);
- ds->data_raw = (pn_bytes_t){0, NULL};
- pn_data_clear(ds->annotations);
- pn_bytes_free(ds->annotations_raw);
- ds->annotations_raw = (pn_bytes_t){0, NULL};
- pn_condition_clear(&ds->condition);
+ pn_disposition_finalize(ds);
+ pn_disposition_init(ds);
}
void pn_delivery_inspect(void *obj, pn_fixed_string_t *dst) {
@@ -1804,7 +1805,7 @@ void pn_delivery_dump(pn_delivery_t *d)
printf("{tag=%s, local.type=%" PRIu64 ", remote.type=%" PRIu64 ",
local.settled=%d, "
"remote.settled=%d, updated=%d, current=%d, writable=%d, readable=%d,
"
"work=%d}",
- tag, d->local.type, d->remote.type, d->local.settled,
+ tag, pn_disposition_type(&d->local), pn_disposition_type(&d->remote),
d->local.settled,
d->remote.settled, d->updated, pn_delivery_current(d),
pn_delivery_writable(d), pn_delivery_readable(d), d->work);
}
@@ -1830,75 +1831,151 @@ pn_record_t *pn_delivery_attachments(pn_delivery_t
*delivery)
uint64_t pn_disposition_type(pn_disposition_t *disposition)
{
assert(disposition);
- return disposition->type;
+ switch (disposition->type) {
+ case PN_DISP_CUSTOM:
+ return disposition->u.s_custom.type;
+ default:
+ // This relies on the disposition types having the protocol values
+ return (uint64_t)disposition->type;
+ }
+}
+
+void pni_disposition_to_raw(pn_disposition_t *disposition) {
+
+ uint64_t type = disposition->type;
+ if (type==PN_DISP_CUSTOM) return;
+
+ char buffer[512];
+ pn_rwbytes_t bytes = {.size=sizeof(buffer), .start=&buffer[0]};
+ pni_emitter_t emitter = make_emitter_from_bytes(bytes);
+ pni_compound_context compound = make_compound();
+
+ switch (type) {
+ case PN_DISP_EMPTY:
+ break;
+ case PN_DISP_RECEIVED:
+ emit_received_disposition(&emitter, &compound,
&disposition->u.s_received);
+ break;
+ case PN_DISP_ACCEPTED:
+ emit_list0(&emitter, &compound);
+ break;
+ case PN_DISP_RELEASED:
+ emit_list0(&emitter, &compound);
+ break;
+ case PN_DISP_REJECTED:
+ emit_rejected_disposition(&emitter, &compound,
&disposition->u.s_rejected);
+ break;
+ case PN_DISP_MODIFIED:
+ emit_modified_disposition(&emitter, &compound,
&disposition->u.s_modified);
+ break;
+ }
+
+ if (type != PN_DISP_EMPTY) {
+ pn_disposition_clear(disposition);
+ disposition->u.s_custom.data_raw =
pn_bytes_dup(make_bytes_from_emitter(emitter));
+ }
+
+ disposition->type = PN_DISP_CUSTOM;
+ disposition->u.s_custom.type = type;
}
pn_data_t *pn_disposition_data(pn_disposition_t *disposition)
{
assert(disposition);
- pni_switch_to_data(&disposition->data_raw, &disposition->data);
- return disposition->data;
+ if (disposition->type != PN_DISP_CUSTOM) {
+ pni_disposition_to_raw(disposition);
+ }
+ pni_switch_to_data(&disposition->u.s_custom.data_raw,
&disposition->u.s_custom.data);
+ return disposition->u.s_custom.data;
}
uint32_t pn_disposition_get_section_number(pn_disposition_t *disposition)
{
assert(disposition);
- return disposition->section_number;
+ if (disposition->type == PN_DISP_RECEIVED) return
disposition->u.s_received.section_number;
+ else return 0;
}
void pn_disposition_set_section_number(pn_disposition_t *disposition, uint32_t
section_number)
{
assert(disposition);
- disposition->section_number = section_number;
+ if (disposition->type != PN_DISP_RECEIVED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_RECEIVED;
+ }
+ disposition->u.s_received.section_number = section_number;
}
uint64_t pn_disposition_get_section_offset(pn_disposition_t *disposition)
{
assert(disposition);
- return disposition->section_offset;
+ if (disposition->type == PN_DISP_RECEIVED) return
disposition->u.s_received.section_offset;
+ else return 0;
}
void pn_disposition_set_section_offset(pn_disposition_t *disposition, uint64_t
section_offset)
{
assert(disposition);
- disposition->section_offset = section_offset;
+ if (disposition->type != PN_DISP_RECEIVED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_RECEIVED;
+ }
+ disposition->u.s_received.section_offset = section_offset;
}
bool pn_disposition_is_failed(pn_disposition_t *disposition)
{
assert(disposition);
- return disposition->failed;
+ if (disposition->type == PN_DISP_MODIFIED) return
disposition->u.s_modified.failed;
+ else return false;
}
void pn_disposition_set_failed(pn_disposition_t *disposition, bool failed)
{
assert(disposition);
- disposition->failed = failed;
+ if (disposition->type != PN_DISP_MODIFIED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_MODIFIED;
+ }
+ disposition->u.s_modified.failed = failed;
}
bool pn_disposition_is_undeliverable(pn_disposition_t *disposition)
{
assert(disposition);
- return disposition->undeliverable;
+ if (disposition->type == PN_DISP_MODIFIED) return
disposition->u.s_modified.undeliverable;
+ else return false;
}
void pn_disposition_set_undeliverable(pn_disposition_t *disposition, bool
undeliverable)
{
assert(disposition);
- disposition->undeliverable = undeliverable;
+ if (disposition->type != PN_DISP_MODIFIED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_MODIFIED;
+ }
+ disposition->u.s_modified.undeliverable = undeliverable;
}
pn_data_t *pn_disposition_annotations(pn_disposition_t *disposition)
{
assert(disposition);
- pni_switch_to_data(&disposition->annotations_raw, &disposition->annotations);
- return disposition->annotations;
+ if (disposition->type != PN_DISP_MODIFIED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_MODIFIED;
+ }
+ pni_switch_to_data(&disposition->u.s_modified.annotations_raw,
&disposition->u.s_modified.annotations);
+ return disposition->u.s_modified.annotations;
}
pn_condition_t *pn_disposition_condition(pn_disposition_t *disposition)
{
assert(disposition);
- return &disposition->condition;
+ if (disposition->type != PN_DISP_REJECTED) {
+ pn_disposition_clear(disposition);
+ disposition->type = PN_DISP_REJECTED;
+ }
+ return &disposition->u.s_rejected.condition;
}
pn_delivery_tag_t pn_delivery_tag(pn_delivery_t *delivery)
@@ -2196,7 +2273,7 @@ pn_disposition_t *pn_delivery_local(pn_delivery_t
*delivery)
uint64_t pn_delivery_local_state(pn_delivery_t *delivery)
{
assert(delivery);
- return delivery->local.type;
+ return pn_disposition_type(&delivery->local);
}
pn_disposition_t *pn_delivery_remote(pn_delivery_t *delivery)
@@ -2208,7 +2285,7 @@ pn_disposition_t *pn_delivery_remote(pn_delivery_t
*delivery)
uint64_t pn_delivery_remote_state(pn_delivery_t *delivery)
{
assert(delivery);
- return delivery->remote.type;
+ return pn_disposition_type(&delivery->remote);
}
bool pn_delivery_settled(pn_delivery_t *delivery)
@@ -2230,7 +2307,34 @@ void pn_delivery_clear(pn_delivery_t *delivery)
void pn_delivery_update(pn_delivery_t *delivery, uint64_t state)
{
if (!delivery) return;
- delivery->local.type = state;
+ if (delivery->local.type == PN_DISP_CUSTOM) {
+ switch (state) {
+ case PN_ACCEPTED:
+ case PN_REJECTED:
+ case PN_RECEIVED:
+ case PN_MODIFIED:
+ case PN_RELEASED:
+ break;
+ default:
+ delivery->local.u.s_custom.type = state;
+ pni_add_tpwork(delivery);
+ return;
+ }
+ }
+ if (delivery->local.type != state) pn_disposition_clear(&delivery->local);
+ switch (state) {
+ case PN_ACCEPTED:
+ case PN_REJECTED:
+ case PN_RECEIVED:
+ case PN_MODIFIED:
+ case PN_RELEASED:
+ delivery->local.type = state;
+ break;
+ default:
+ delivery->local.type = PN_DISP_CUSTOM;
+ delivery->local.u.s_custom.type = state;
+ break;
+ }
pni_add_tpwork(delivery);
}
diff --git a/c/src/core/transport.c b/c/src/core/transport.c
index 1fced2f84..892e430c1 100644
--- a/c/src/core/transport.c
+++ b/c/src/core/transport.c
@@ -1331,6 +1331,8 @@ static void pn_full_settle(pn_delivery_map_t *db,
pn_delivery_t *delivery)
pn_decref(delivery);
}
+static void pni_amqp_decode_disposition (uint64_t type, pn_bytes_t disp_data,
pn_disposition_t *disp);
+
int pn_do_transfer(pn_transport_t *transport, uint8_t frame_type, uint16_t
channel, pn_bytes_t payload)
{
// XXX: multi transfer
@@ -1414,9 +1416,7 @@ int pn_do_transfer(pn_transport_t *transport, uint8_t
frame_type, uint16_t chann
state->id, id);
}
if (has_type) {
- delivery->remote.type = type;
- pn_bytes_free(delivery->remote.data_raw);
- delivery->remote.data_raw = pn_bytes_dup(disp_data);
+ pni_amqp_decode_disposition(type, disp_data, &delivery->remote);
}
link->state.delivery_count++;
link->state.link_credit--;
@@ -1546,70 +1546,80 @@ static inline bool sequence_lte(pn_sequence_t a,
pn_sequence_t b) {
return b-a <= INT32_MAX;
}
-static int pni_do_delivery_disposition(pn_transport_t * transport,
pn_delivery_t *delivery, bool settled, bool remote_data, bool type_init,
uint64_t type, pn_bytes_t disp_data) {
- pn_disposition_t *remote = &delivery->remote;
-
- if (type_init) remote->type = type;
+static void pni_amqp_decode_disposition (uint64_t type, pn_bytes_t disp_data,
pn_disposition_t *disp) {
+ switch (type) {
+ case AMQP_DESC_RECEIVED: {
+ bool qnumber;
+ uint32_t number;
+ bool qoffset;
+ uint64_t offset;
+ pn_amqp_decode_DqEQIQLe(disp_data, &qnumber, &number, &qoffset, &offset);
- if (remote_data) {
- switch (type) {
- case PN_RECEIVED: {
- bool qnumber;
- uint32_t number;
- bool qoffset;
- uint64_t offset;
- pn_amqp_decode_DqEQIQLe(disp_data, &qnumber, &number, &qoffset,
&offset);
+ disp->type = PN_DISP_RECEIVED;
- if (qnumber) {
- remote->section_number = number;
- }
- if (qoffset) {
- remote->section_offset = offset;
- }
- break;
+ if (qnumber) {
+ disp->u.s_received.section_number = number;
+ }
+ if (qoffset) {
+ disp->u.s_received.section_offset = offset;
}
- case PN_ACCEPTED:
- break;
+ break;
+ }
+ case AMQP_DESC_ACCEPTED:
+ disp->type = PN_DISP_ACCEPTED;
+ break;
- case PN_REJECTED: {
- pn_bytes_t cond;
- pn_bytes_t desc;
- pn_bytes_t info;
- pn_amqp_decode_DqEDqEsSRee(disp_data, &cond, &desc, &info);
- pn_condition_set(&remote->condition, cond, desc, info);
+ case AMQP_DESC_REJECTED: {
+ pn_bytes_t cond;
+ pn_bytes_t desc;
+ pn_bytes_t info;
+ pn_amqp_decode_DqEDqEsSRee(disp_data, &cond, &desc, &info);
+ disp->type = PN_DISP_REJECTED;
+ pn_condition_set(&disp->u.s_rejected.condition, cond, desc, info);
- break;
- }
- case PN_RELEASED:
- break;
-
- case PN_MODIFIED: {
- bool qfailed;
- bool failed;
- bool qundeliverable;
- bool undeliverable;
- pn_bytes_t annotations_raw = (pn_bytes_t){0, NULL};
- pn_amqp_decode_DqEQoQoRe(disp_data, &qfailed, &failed,
&qundeliverable, &undeliverable, &annotations_raw);
- pn_bytes_free(remote->annotations_raw);
- remote->annotations_raw = pn_bytes_dup(annotations_raw);
-
- if (qfailed) {
- remote->failed = failed;
- }
- if (qundeliverable) {
- remote->undeliverable = undeliverable;
- }
- break;
+ break;
+ }
+ case AMQP_DESC_RELEASED:
+ disp->type = PN_DISP_RELEASED;
+ break;
+
+ case AMQP_DESC_MODIFIED: {
+ bool qfailed;
+ bool failed;
+ bool qundeliverable;
+ bool undeliverable;
+ pn_bytes_t annotations_raw = (pn_bytes_t){0, NULL};
+ pn_amqp_decode_DqEQoQoRe(disp_data, &qfailed, &failed, &qundeliverable,
&undeliverable, &annotations_raw);
+ disp->type = PN_DISP_MODIFIED;
+ pn_bytes_free(disp->u.s_modified.annotations_raw);
+ disp->u.s_modified.annotations_raw = pn_bytes_dup(annotations_raw);
+
+ if (qfailed) {
+ disp->u.s_modified.failed = failed;
}
- default: {
- pn_bytes_t data_raw = (pn_bytes_t){0, NULL};
- pn_amqp_decode_DqR(disp_data, &data_raw);
- pn_bytes_free(remote->data_raw);
- remote->data_raw = pn_bytes_dup(data_raw);
- break;
+ if (qundeliverable) {
+ disp->u.s_modified.undeliverable = undeliverable;
}
+ break;
+ }
+ default: {
+ pn_bytes_t data_raw = (pn_bytes_t){0, NULL};
+ pn_amqp_decode_DqR(disp_data, &data_raw);
+ disp->type = PN_DISP_CUSTOM;
+ disp->u.s_custom.type = type;
+ pn_bytes_free(disp->u.s_custom.data_raw);
+ disp->u.s_custom.data_raw = pn_bytes_dup(data_raw);
+ break;
}
}
+}
+
+static int pni_do_delivery_disposition(pn_transport_t * transport,
pn_delivery_t *delivery, bool settled, bool remote_data, bool type_init,
uint64_t type, pn_bytes_t disp_data) {
+ pn_disposition_t *remote = &delivery->remote;
+
+ if (remote_data && type_init) {
+ pni_amqp_decode_disposition(type, disp_data, remote);
+ }
remote->settled = settled;
delivery->updated = true;
diff --git a/python/proton/_delivery.py b/python/proton/_delivery.py
index 6332fd8a7..1929d82bc 100644
--- a/python/proton/_delivery.py
+++ b/python/proton/_delivery.py
@@ -323,9 +323,12 @@ class Delivery(Wrapper):
:param state: State of delivery
"""
- obj2dat(self.local._data, pn_disposition_data(self.local._impl))
- obj2dat(self.local._annotations,
pn_disposition_annotations(self.local._impl))
- obj2cond(self.local._condition,
pn_disposition_condition(self.local._impl))
+ if state == self.MODIFIED:
+ obj2dat(self.local._annotations,
pn_disposition_annotations(self.local._impl))
+ elif state == self.REJECTED:
+ obj2cond(self.local._condition,
pn_disposition_condition(self.local._impl))
+ elif state not in (self.ACCEPTED, self.RECEIVED, self.RELEASED):
+ obj2dat(self.local._data, pn_disposition_data(self.local._impl))
pn_delivery_update(self._impl, state)
@property
diff --git a/python/tests/proton_tests/engine.py
b/python/tests/proton_tests/engine.py
index fc98a7056..e753bae58 100644
--- a/python/tests/proton_tests/engine.py
+++ b/python/tests/proton_tests/engine.py
@@ -2271,13 +2271,6 @@ class DispositionTester:
def check(self, dlv: Delivery):
assert dlv.remote_state == self._type
assert dlv.remote.type == self._type
- assert dlv.remote.data is None
- assert dlv.remote.section_number == 0
- assert dlv.remote.section_offset == 0
- assert dlv.remote.condition is None
- assert dlv.remote.failed is False
- assert dlv.remote.undeliverable is False
- assert dlv.remote.annotations is None
class RejectedTester(DispositionTester):
@@ -2292,13 +2285,7 @@ class RejectedTester(DispositionTester):
def check(self, dlv: Delivery):
assert dlv.remote_state == self._type
assert dlv.remote.type == self._type
- assert dlv.remote.data is None, dlv.data
- assert dlv.remote.section_number == 0
- assert dlv.remote.section_offset == 0
assert dlv.remote.condition == self.condition, (dlv.condition,
self.condition)
- assert dlv.remote.failed is False
- assert dlv.remote.undeliverable is False
- assert dlv.remote.annotations is None
class ReceivedValue(DispositionTester):
@@ -2315,13 +2302,8 @@ class ReceivedValue(DispositionTester):
def check(self, dlv: Delivery):
assert dlv.remote_state == self._type
assert dlv.remote.type == self._type
- assert dlv.remote.data is None, dlv.data
assert dlv.remote.section_number == self._section_number,
(dlv.section_number, self._section_number)
assert dlv.remote.section_offset == self._section_offset
- assert dlv.remote.condition is None
- assert dlv.remote.failed is False
- assert dlv.remote.undeliverable is False
- assert dlv.remote.annotations is None
class ModifiedTester(DispositionTester):
@@ -2340,10 +2322,6 @@ class ModifiedTester(DispositionTester):
def check(self, dlv: Delivery):
assert dlv.remote_state == self._type
assert dlv.remote.type == self._type
- assert dlv.remote.data is None, dlv.data
- assert dlv.remote.section_number == 0
- assert dlv.remote.section_offset == 0
- assert dlv.remote.condition is None
assert dlv.remote.failed == self._failed
assert dlv.remote.undeliverable == self._undeliverable
assert dlv.remote.annotations == self._annotations, (dlv.annotations,
self._annotations)
@@ -2362,12 +2340,6 @@ class CustomTester(DispositionTester):
assert dlv.remote_state == self._type
assert dlv.remote.type == self._type
assert dlv.remote.data == self._data, (dlv.data, self._data)
- assert dlv.remote.section_number == 0
- assert dlv.remote.section_offset == 0
- assert dlv.remote.condition is None
- assert dlv.remote.failed is False
- assert dlv.remote.undeliverable is False
- assert dlv.remote.annotations is None
class DeliveryTest(Test):
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org
