You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rh...@apache.org on 2012/10/11 19:15:15 UTC
svn commit: r1397174 - in /qpid/proton/trunk: proton-c/bindings/python/
proton-c/include/proton/ proton-c/src/ proton-c/src/codec/
proton-c/src/engine/ proton-c/src/message/ proton-j/src/main/scripts/
tests/proton_tests/
Author: rhs
Date: Thu Oct 11 17:15:14 2012
New Revision: 1397174
URL: http://svn.apache.org/viewvc?rev=1397174&view=rev
Log:
full support for source and target and message sections
Modified:
qpid/proton/trunk/proton-c/bindings/python/proton.py
qpid/proton/trunk/proton-c/bindings/python/python.i
qpid/proton/trunk/proton-c/include/proton/codec.h
qpid/proton/trunk/proton-c/include/proton/engine.h
qpid/proton/trunk/proton-c/include/proton/message.h
qpid/proton/trunk/proton-c/include/proton/types.h
qpid/proton/trunk/proton-c/src/codec/codec.c
qpid/proton/trunk/proton-c/src/engine/engine-internal.h
qpid/proton/trunk/proton-c/src/engine/engine.c
qpid/proton/trunk/proton-c/src/message/message.c
qpid/proton/trunk/proton-c/src/messenger.c
qpid/proton/trunk/proton-c/src/proton.c
qpid/proton/trunk/proton-j/src/main/scripts/proton.py
qpid/proton/trunk/tests/proton_tests/codec.py
qpid/proton/trunk/tests/proton_tests/engine.py
Modified: qpid/proton/trunk/proton-c/bindings/python/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/bindings/python/proton.py?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/bindings/python/proton.py (original)
+++ qpid/proton/trunk/proton-c/bindings/python/proton.py Thu Oct 11 17:15:14 2012
@@ -595,6 +595,22 @@ The format of the message.
self._check(err)
return data
+ @property
+ def instructions(self):
+ return Data(_data=pn_message_instructions(self._msg))
+
+ @property
+ def annotations(self):
+ return Data(_data=pn_message_annotations(self._msg))
+
+ @property
+ def properties(self):
+ return Data(_data=pn_message_properties(self._msg))
+
+ @property
+ def body(self):
+ return Data(_data=pn_message_body(self._msg))
+
class DataException(ProtonException):
"""
The DataException class is the root of the Data exception hierarchy.
@@ -680,11 +696,16 @@ class Data:
LIST = PN_LIST; "A list value."
MAP = PN_MAP; "A map value."
- def __init__(self, capacity=16):
- self._data = pn_data(capacity)
+ def __init__(self, capacity=16, _data=None):
+ if _data:
+ self._data = _data or pn_data(capacity)
+ self._free = False
+ else:
+ self._data = pn_data(capacity)
+ self._free = True
def __del__(self):
- if hasattr(self, "_data"):
+ if self._free and hasattr(self, "_data"):
pn_data_free(self._data)
del self._data
@@ -1244,6 +1265,20 @@ class Data:
"""
return pn_data_get_symbol(self._data)
+ def copy(self, src):
+ self._check(pn_data_copy(self._data, src._data))
+
+ def format(self):
+ sz = 16
+ while True:
+ err, result = pn_data_format(self._data, sz)
+ if err == PN_OVERFLOW:
+ sz *= 2
+ continue
+ else:
+ self._check(err)
+ return result
+
def dump(self):
pn_data_dump(self._data)
@@ -1422,24 +1457,20 @@ class Link(Endpoint):
def state(self):
return pn_link_state(self._link)
- def _set_source(self, source):
- pn_link_set_source(self._link, source)
- def _get_source(self):
- return pn_link_get_source(self._link)
- source = property(_get_source, _set_source)
-
- def _set_target(self, target):
- pn_link_set_target(self._link, target)
- def _get_target(self):
- return pn_link_get_target(self._link)
- target = property(_get_target, _set_target)
+ @property
+ def source(self):
+ return Terminus(pn_link_source(self._link))
+
+ @property
+ def target(self):
+ return Terminus(pn_link_target(self._link))
@property
def remote_source(self):
- return pn_link_remote_source(self._link)
+ return Terminus(pn_link_remote_source(self._link))
@property
def remote_target(self):
- return pn_link_remote_target(self._link)
+ return Terminus(pn_link_remote_target(self._link))
@property
def session(self):
@@ -1474,6 +1505,83 @@ class Link(Endpoint):
def next(self, mask):
return wrap_link(pn_link_next(self._link, mask))
+class Terminus(object):
+
+ UNSPECIFIED = PN_UNSPECIFIED
+ SOURCE = PN_SOURCE
+ TARGET = PN_TARGET
+ COORDINATOR = PN_COORDINATOR
+
+ NONDURABLE = PN_NONDURABLE
+ CONFIGURATION = PN_CONFIGURATION
+ DELIVERIES = PN_DELIVERIES
+
+ def __init__(self, impl):
+ self._impl = impl
+
+ def _check(self, err):
+ if err < 0:
+ exc = EXCEPTIONS.get(err, LinkException)
+ raise exc("[%s]" % err)
+ else:
+ return err
+
+ def _get_type(self):
+ return pn_terminus_get_type(self._impl)
+ def _set_type(self, type):
+ self._check(pn_terminus_set_type(self._impl, type))
+ type = property(_get_type, _set_type)
+
+ def _get_address(self):
+ return pn_terminus_get_address(self._impl)
+ def _set_address(self, address):
+ self._check(pn_terminus_set_address(self._impl, address))
+ address = property(_get_address, _set_address)
+
+ def _get_durability(self):
+ return pn_terminus_get_durability(self._impl)
+ def _set_durability(self, seconds):
+ self._check(pn_terminus_set_durability(self._impl, seconds))
+ durability = property(_get_durability, _set_durability)
+
+ def _get_expiry_policy(self):
+ return pn_terminus_get_expiry_policy(self._impl)
+ def _set_expiry_policy(self, seconds):
+ self._check(pn_terminus_set_expiry_policy(self._impl, seconds))
+ expiry_policy = property(_get_expiry_policy, _set_expiry_policy)
+
+ def _get_timeout(self):
+ return pn_terminus_get_timeout(self._impl)
+ def _set_timeout(self, seconds):
+ self._check(pn_terminus_set_timeout(self._impl, seconds))
+ timeout = property(_get_timeout, _set_timeout)
+
+ def _is_dynamic(self):
+ return pn_terminus_is_dynamic(self._impl)
+ def _set_dynamic(self, dynamic):
+ self._check(pn_terminus_set_dynamic(self._impl, dynamic))
+ dynamic = property(_is_dynamic, _set_dynamic)
+
+ @property
+ def properties(self):
+ return Data(_data = pn_terminus_properties(self._impl))
+
+ @property
+ def capabilities(self):
+ return Data(_data = pn_terminus_capabilities(self._impl))
+
+ @property
+ def outcomes(self):
+ return Data(_data = pn_terminus_outcomes(self._impl))
+
+ @property
+ def filter(self):
+ return Data(_data = pn_terminus_filter(self._impl))
+
+ def copy(self, src):
+ self._check(pn_terminus_copy(self._impl, src._impl))
+
+
class Sender(Link):
def offered(self, n):
@@ -1705,5 +1813,6 @@ class SSL(object):
__all__ = ["Messenger", "Message", "ProtonException", "MessengerException",
"MessageException", "Timeout", "Data", "Endpoint", "Connection",
- "Session", "Link", "Sender", "Receiver", "Delivery", "Transport",
- "TransportException", "SASL", "SSL", "PN_SESSION_WINDOW"]
+ "Session", "Link", "Terminus", "Sender", "Receiver", "Delivery",
+ "Transport", "TransportException", "SASL", "SSL",
+ "PN_SESSION_WINDOW"]
Modified: qpid/proton/trunk/proton-c/bindings/python/python.i
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/bindings/python/python.i?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/bindings/python/python.i (original)
+++ qpid/proton/trunk/proton-c/bindings/python/python.i Thu Oct 11 17:15:14 2012
@@ -390,5 +390,7 @@ ssize_t pn_data_decode(pn_data_t *data,
%}
%ignore pn_sasl_recv;
+int pn_data_format(pn_data_t *data, char *OUTPUT, size_t *OUTPUT_SIZE);
+%ignore pn_data_format;
%include "proton/cproton.i"
Modified: qpid/proton/trunk/proton-c/include/proton/codec.h
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/include/proton/codec.h?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/include/proton/codec.h (original)
+++ qpid/proton/trunk/proton-c/include/proton/codec.h Thu Oct 11 17:15:14 2012
@@ -171,6 +171,12 @@ pn_bytes_t pn_data_get_binary(pn_data_t
pn_bytes_t pn_data_get_string(pn_data_t *data);
pn_bytes_t pn_data_get_symbol(pn_data_t *data);
+int pn_data_copy(pn_data_t *data, pn_data_t *src);
+int pn_data_append(pn_data_t *data, pn_data_t *src);
+int pn_data_appendn(pn_data_t *data, pn_data_t *src, int limit);
+void pn_data_narrow(pn_data_t *data);
+void pn_data_widen(pn_data_t *data);
+
void pn_data_dump(pn_data_t *data);
#ifdef __cplusplus
Modified: qpid/proton/trunk/proton-c/include/proton/engine.h
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/include/proton/engine.h?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/include/proton/engine.h (original)
+++ qpid/proton/trunk/proton-c/include/proton/engine.h Thu Oct 11 17:15:14 2012
@@ -25,6 +25,7 @@
#include <stdbool.h>
#include <stddef.h>
#include <sys/types.h>
+#include <proton/codec.h>
#include <proton/error.h>
#ifdef __cplusplus
@@ -41,6 +42,24 @@ typedef struct pn_transport_t pn_transpo
typedef struct pn_connection_t pn_connection_t; /**< Connection */
typedef struct pn_session_t pn_session_t; /**< Session */
typedef struct pn_link_t pn_link_t; /**< Link */
+typedef struct pn_terminus_t pn_terminus_t;
+typedef enum {
+ PN_UNSPECIFIED = 0,
+ PN_SOURCE = 1,
+ PN_TARGET = 2,
+ PN_COORDINATOR = 3
+} pn_terminus_type_t;
+typedef enum {
+ PN_NONDURABLE = 0,
+ PN_CONFIGURATION = 1,
+ PN_DELIVERIES = 2
+} pn_durability_t;
+typedef enum {
+ PN_LINK_CLOSE,
+ PN_SESSION_CLOSE,
+ PN_CONNECTION_CLOSE,
+ PN_NEVER,
+} pn_expiry_policy_t;
typedef struct pn_delivery_t pn_delivery_t;
typedef struct pn_delivery_tag_t {
@@ -276,12 +295,10 @@ bool pn_link_is_receiver(pn_link_t *link
pn_state_t pn_link_state(pn_link_t *link);
pn_error_t *pn_link_error(pn_link_t *link);
pn_session_t *pn_link_session(pn_link_t *link);
-const char *pn_link_get_target(pn_link_t *link);
-const char *pn_link_get_source(pn_link_t *link);
-void pn_link_set_source(pn_link_t *link, const char *source);
-void pn_link_set_target(pn_link_t *link, const char *target);
-const char *pn_link_remote_source(pn_link_t *link);
-const char *pn_link_remote_target(pn_link_t *link);
+pn_terminus_t *pn_link_source(pn_link_t *link);
+pn_terminus_t *pn_link_target(pn_link_t *link);
+pn_terminus_t *pn_link_remote_source(pn_link_t *link);
+pn_terminus_t *pn_link_remote_target(pn_link_t *link);
pn_delivery_t *pn_link_current(pn_link_t *link);
bool pn_link_advance(pn_link_t *link);
int pn_link_credit(pn_link_t *link);
@@ -309,6 +326,27 @@ void pn_link_flow(pn_link_t *receiver, i
void pn_link_drain(pn_link_t *receiver, int credit);
ssize_t pn_link_recv(pn_link_t *receiver, char *bytes, size_t n);
+// terminus
+pn_terminus_type_t pn_terminus_get_type(pn_terminus_t *terminus);
+int pn_terminus_set_type(pn_terminus_t *terminus, pn_terminus_type_t type);
+
+const char *pn_terminus_get_address(pn_terminus_t *terminus);
+int pn_terminus_set_address(pn_terminus_t *terminus, const char *address);
+pn_durability_t pn_terminus_get_durability(pn_terminus_t *terminus);
+int pn_terminus_set_durability(pn_terminus_t *terminus,
+ pn_durability_t durability);
+pn_expiry_policy_t pn_terminus_get_expiry_policy(pn_terminus_t *terminus);
+int pn_terminus_set_expiry_policy(pn_terminus_t *terminus, pn_expiry_policy_t policy);
+pn_seconds_t pn_terminus_get_timeout(pn_terminus_t *terminus);
+int pn_terminus_set_timeout(pn_terminus_t *terminus, pn_seconds_t);
+bool pn_terminus_is_dynamic(pn_terminus_t *terminus);
+int pn_terminus_set_dynamic(pn_terminus_t *terminus, bool dynamic);
+pn_data_t *pn_terminus_properties(pn_terminus_t *terminus);
+pn_data_t *pn_terminus_capabilities(pn_terminus_t *terminus);
+pn_data_t *pn_terminus_outcomes(pn_terminus_t *terminus);
+pn_data_t *pn_terminus_filter(pn_terminus_t *terminus);
+int pn_terminus_copy(pn_terminus_t *terminus, pn_terminus_t *src);
+
// delivery
pn_delivery_t *pn_delivery(pn_link_t *link, pn_delivery_tag_t tag);
pn_delivery_tag_t pn_delivery_tag(pn_delivery_t *delivery);
Modified: qpid/proton/trunk/proton-c/include/proton/message.h
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/include/proton/message.h?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/include/proton/message.h (original)
+++ qpid/proton/trunk/proton-c/include/proton/message.h Thu Oct 11 17:15:14 2012
@@ -103,22 +103,6 @@ int pn_message_set_group_sequ
const char * pn_message_get_reply_to_group_id (pn_message_t *msg);
int pn_message_set_reply_to_group_id (pn_message_t *msg, const char *reply_to_group_id);
-/* path based getters/setters, allows access to delivery-annotations,
- message-annotations, application-properties, footer */
-
-/*
-uint32_t pn_message_get_uint32 (pn_message_t *msg, const char *path);
-int pn_message_set_uint32 (pn_message_t *msg, const char *path, uint32_t value);
-
-// ...
-
-int pn_message_get_data (pn_message_t *msg, const char *path, pn_data_t *data);
-int pn_message_set_data (pn_message_t *msg, const char *path, pn_data_t *data);
-
-int pn_message_get_json (pn_message_t *msg, const char *path, char *bytes, size_t *size);
-int pn_message_set_json (pn_message_t *msg, const char *path, const char *bytes);
-*/
-
pn_format_t pn_message_get_format(pn_message_t *message);
int pn_message_set_format(pn_message_t *message, pn_format_t format);
@@ -134,6 +118,11 @@ int pn_message_save_text(pn_message_t *m
int pn_message_save_amqp(pn_message_t *message, char *data, size_t *size);
int pn_message_save_json(pn_message_t *message, char *data, size_t *size);
+pn_data_t *pn_message_instructions(pn_message_t *msg);
+pn_data_t *pn_message_annotations(pn_message_t *msg);
+pn_data_t *pn_message_properties(pn_message_t *msg);
+pn_data_t *pn_message_body(pn_message_t *msg);
+
// TODO:
// bind vars
//int pn_message_set(pn_message_t *message, int i, pn_atom_t *atom);
Modified: qpid/proton/trunk/proton-c/include/proton/types.h
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/include/proton/types.h?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/include/proton/types.h (original)
+++ qpid/proton/trunk/proton-c/include/proton/types.h Thu Oct 11 17:15:14 2012
@@ -31,6 +31,7 @@ extern "C" {
typedef int32_t pn_sequence_t;
typedef uint32_t pn_millis_t;
+typedef uint32_t pn_seconds_t;
typedef uint64_t pn_timestamp_t;
typedef uint32_t pn_char_t;
typedef uint32_t pn_decimal32_t;
Modified: qpid/proton/trunk/proton-c/src/codec/codec.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/codec/codec.c?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/codec/codec.c (original)
+++ qpid/proton/trunk/proton-c/src/codec/codec.c Thu Oct 11 17:15:14 2012
@@ -176,7 +176,7 @@ int pn_format_atom(pn_bytes_t *bytes, pn
case PN_DECIMAL64:
return pn_bytes_format(bytes, "D64(%" PRIu64 ")", atom.u.as_decimal64);
case PN_DECIMAL128:
- return pn_bytes_format(bytes, "D128(%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x%x)",
+ return pn_bytes_format(bytes, "D128(%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx)",
atom.u.as_decimal128.bytes[0],
atom.u.as_decimal128.bytes[1],
atom.u.as_decimal128.bytes[2],
@@ -194,7 +194,9 @@ int pn_format_atom(pn_bytes_t *bytes, pn
atom.u.as_decimal128.bytes[14],
atom.u.as_decimal128.bytes[15]);
case PN_UUID:
- return pn_bytes_format(bytes, "UUID(%x%x%x%x-%x%x-%x%x-%x%x-%x%x%x%x%x%x)",
+ return pn_bytes_format(bytes, "UUID(%02hhx%02hhx%02hhx%02hhx-"
+ "%02hhx%02hhx-%02hhx%02hhx-%02hhx%02hhx-"
+ "%02hhx%02hhx%02hhx%02hhx%02hhx%02hhx)",
atom.u.as_uuid.bytes[0],
atom.u.as_uuid.bytes[1],
atom.u.as_uuid.bytes[2],
@@ -1834,10 +1836,17 @@ struct pn_data_t {
pn_buffer_t *buf;
size_t parent;
size_t current;
+ size_t base_parent;
+ size_t base_current;
size_t extras;
pn_error_t *error;
};
+typedef struct {
+ size_t parent;
+ size_t current;
+} pn_point_t;
+
pn_data_t *pn_data(size_t capacity)
{
pn_data_t *data = malloc(sizeof(pn_data_t));
@@ -1847,6 +1856,8 @@ pn_data_t *pn_data(size_t capacity)
data->buf = pn_buffer(64);
data->parent = 0;
data->current = 0;
+ data->base_parent = 0;
+ data->base_current = 0;
data->extras = 0;
data->error = pn_error();
return data;
@@ -1884,6 +1895,8 @@ void pn_data_clear(pn_data_t *data)
data->extras = 0;
data->parent = 0;
data->current = 0;
+ data->base_parent = 0;
+ data->base_current = 0;
pn_buffer_clear(data->buf);
}
}
@@ -2097,6 +2110,18 @@ int pn_data_vfill(pn_data_t *data, const
}
}
break;
+ case 'C':
+ {
+ pn_data_t *src = va_arg(ap, pn_data_t *);
+ if (src && pn_data_size(src) > 0) {
+ err = pn_data_appendn(data, src, 1);
+ if (err) return err;
+ } else {
+ err = pn_data_put_null(data);
+ if (err) return err;
+ }
+ }
+ break;
default:
fprintf(stderr, "unrecognized fill code: 0x%.2X '%c'\n", code, code);
return PN_ARG_ERR;
@@ -2154,6 +2179,8 @@ static bool pn_scan_next(pn_data_t *data
}
}
+pn_node_t *pn_data_peek(pn_data_t *data);
+
int pn_data_vscan(pn_data_t *data, const char *fmt, va_list ap)
{
pn_data_rewind(data);
@@ -2495,6 +2522,28 @@ int pn_data_vscan(pn_data_t *data, const
return pn_error_format(data->error, PN_ARG_ERR, "codes must follow a ?");
scanarg = va_arg(ap, bool *);
break;
+ case 'C':
+ {
+ pn_data_t *dst = va_arg(ap, pn_data_t *);
+ if (!suspend) {
+ size_t old = pn_data_size(dst);
+ pn_node_t *next = pn_data_peek(data);
+ if (next && next->atom.type != PN_NULL) {
+ pn_data_narrow(data);
+ int err = pn_data_appendn(dst, data, 1);
+ pn_data_widen(data);
+ if (err) return err;
+ scanned = pn_data_size(dst) > old;
+ } else {
+ scanned = false;
+ }
+ pn_data_next(data);
+ } else {
+ scanned = false;
+ }
+ }
+ if (resume_count && level == count_level) resume_count--;
+ break;
default:
return pn_error_format(data->error, PN_ARG_ERR, "unrecognized scan code: 0x%.2X '%c'", code, code);
}
@@ -2578,8 +2627,8 @@ pn_node_t *pn_data_new(pn_data_t *data)
void pn_data_rewind(pn_data_t *data)
{
- data->parent = 0;
- data->current = 0;
+ data->parent = data->base_parent;
+ data->current = data->base_current;
}
pn_node_t *pn_data_current(pn_data_t *data)
@@ -2587,6 +2636,52 @@ pn_node_t *pn_data_current(pn_data_t *da
return pn_data_node(data, data->current);
}
+void pn_data_narrow(pn_data_t *data)
+{
+ data->base_parent = data->parent;
+ data->base_current = data->current;
+}
+
+void pn_data_widen(pn_data_t *data)
+{
+ data->base_parent = 0;
+ data->base_current = 0;
+}
+
+pn_point_t pn_data_point(pn_data_t *data)
+{
+ pn_point_t point;
+ point.parent = data->parent;
+ point.current = data->current;
+ return point;
+}
+
+bool pn_data_restore(pn_data_t *data, pn_point_t point)
+{
+ if (point.current && point.current <= data->size) {
+ data->current = point.current;
+ pn_node_t *current = pn_data_current(data);
+ data->parent = current->parent;
+ return true;
+ } else if (point.parent && point.parent <= data->size) {
+ data->parent = point.parent;
+ data->current = 0;
+ return true;
+ } else {
+ return false;
+ }
+}
+
+pn_node_t *pn_data_peek(pn_data_t *data)
+{
+ pn_node_t *current = pn_data_current(data);
+ if (current) {
+ return pn_data_node(data, current->next);
+ } else {
+ return NULL;
+ }
+}
+
bool pn_data_next(pn_data_t *data)
{
pn_node_t *current = pn_data_current(data);
@@ -3460,3 +3555,174 @@ pn_bytes_t pn_data_get_symbol(pn_data_t
return (pn_bytes_t) {0};
}
}
+
+int pn_data_copy(pn_data_t *data, pn_data_t *src)
+{
+ pn_data_clear(data);
+ int err = pn_data_append(data, src);
+ pn_data_rewind(data);
+ return err;
+}
+
+int pn_data_append(pn_data_t *data, pn_data_t *src)
+{
+ return pn_data_appendn(data, src, -1);
+}
+
+int pn_data_appendn(pn_data_t *data, pn_data_t *src, int limit)
+{
+ int err;
+ int level = 0, count = 0;
+ bool stop = false;
+ pn_point_t point = pn_data_point(src);
+ pn_data_rewind(src);
+
+ while (true) {
+ while (!pn_data_next(src)) {
+ if (level > 0) {
+ pn_data_exit(data);
+ pn_data_exit(src);
+ level--;
+ }
+
+ if (!pn_data_next(src)) {
+ stop = true;
+ break;
+ }
+ }
+
+ if (stop) break;
+
+ if (level == 0 && count == limit)
+ break;
+
+ pn_type_t type = pn_data_type(src);
+ switch (type) {
+ case PN_NULL:
+ err = pn_data_put_null(data);
+ if (level == 0) count++;
+ break;
+ case PN_BOOL:
+ err = pn_data_put_bool(data, pn_data_get_bool(src));
+ if (level == 0) count++;
+ break;
+ case PN_UBYTE:
+ err = pn_data_put_ubyte(data, pn_data_get_ubyte(src));
+ if (level == 0) count++;
+ break;
+ case PN_BYTE:
+ err = pn_data_put_byte(data, pn_data_get_byte(src));
+ if (level == 0) count++;
+ break;
+ case PN_USHORT:
+ err = pn_data_put_ushort(data, pn_data_get_ushort(src));
+ if (level == 0) count++;
+ break;
+ case PN_SHORT:
+ err = pn_data_put_short(data, pn_data_get_short(src));
+ if (level == 0) count++;
+ break;
+ case PN_UINT:
+ err = pn_data_put_uint(data, pn_data_get_uint(src));
+ if (level == 0) count++;
+ break;
+ case PN_INT:
+ err = pn_data_put_int(data, pn_data_get_int(src));
+ if (level == 0) count++;
+ break;
+ case PN_CHAR:
+ err = pn_data_put_char(data, pn_data_get_char(src));
+ if (level == 0) count++;
+ break;
+ case PN_ULONG:
+ err = pn_data_put_ulong(data, pn_data_get_ulong(src));
+ if (level == 0) count++;
+ break;
+ case PN_LONG:
+ err = pn_data_put_long(data, pn_data_get_long(src));
+ if (level == 0) count++;
+ break;
+ case PN_TIMESTAMP:
+ err = pn_data_put_timestamp(data, pn_data_get_timestamp(src));
+ if (level == 0) count++;
+ break;
+ case PN_FLOAT:
+ err = pn_data_put_float(data, pn_data_get_float(src));
+ if (level == 0) count++;
+ break;
+ case PN_DOUBLE:
+ err = pn_data_put_double(data, pn_data_get_double(src));
+ if (level == 0) count++;
+ break;
+ case PN_DECIMAL32:
+ err = pn_data_put_decimal32(data, pn_data_get_decimal32(src));
+ if (level == 0) count++;
+ break;
+ case PN_DECIMAL64:
+ err = pn_data_put_decimal64(data, pn_data_get_decimal64(src));
+ if (level == 0) count++;
+ break;
+ case PN_DECIMAL128:
+ err = pn_data_put_decimal128(data, pn_data_get_decimal128(src));
+ if (level == 0) count++;
+ break;
+ case PN_UUID:
+ err = pn_data_put_uuid(data, pn_data_get_uuid(src));
+ if (level == 0) count++;
+ break;
+ case PN_BINARY:
+ err = pn_data_put_binary(data, pn_data_get_binary(src));
+ if (level == 0) count++;
+ break;
+ case PN_STRING:
+ err = pn_data_put_string(data, pn_data_get_string(src));
+ if (level == 0) count++;
+ break;
+ case PN_SYMBOL:
+ err = pn_data_put_symbol(data, pn_data_get_symbol(src));
+ if (level == 0) count++;
+ break;
+ case PN_DESCRIPTOR:
+ err = pn_data_put_described(data);
+ if (level == 0) count++;
+ if (err) { pn_data_restore(src, point); return err; }
+ pn_data_enter(data);
+ pn_data_enter(src);
+ level++;
+ break;
+ case PN_ARRAY:
+ err = pn_data_put_array(data, pn_data_is_array_described(src),
+ pn_data_get_array_type(src));
+ if (level == 0) count++;
+ if (err) { pn_data_restore(src, point); return err; }
+ pn_data_enter(data);
+ pn_data_enter(src);
+ level++;
+ break;
+ case PN_LIST:
+ err = pn_data_put_list(data);
+ if (level == 0) count++;
+ if (err) { pn_data_restore(src, point); return err; }
+ pn_data_enter(data);
+ pn_data_enter(src);
+ level++;
+ break;
+ case PN_MAP:
+ err = pn_data_put_map(data);
+ if (level == 0) count++;
+ if (err) { pn_data_restore(src, point); return err; }
+ pn_data_enter(data);
+ pn_data_enter(src);
+ level++;
+ break;
+ case PN_TYPE:
+ return PN_ERR;
+ }
+
+ if (err) { pn_data_restore(src, point); return err; }
+ }
+
+ pn_data_restore(src, point);
+
+ return 0;
+}
Modified: qpid/proton/trunk/proton-c/src/engine/engine-internal.h
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/engine/engine-internal.h?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/engine/engine-internal.h (original)
+++ qpid/proton/trunk/proton-c/src/engine/engine-internal.h Thu Oct 11 17:15:14 2012
@@ -141,14 +141,27 @@ struct pn_session_t {
void *context;
};
+struct pn_terminus_t {
+ pn_terminus_type_t type;
+ char *address;
+ pn_durability_t durability;
+ pn_expiry_policy_t expiry_policy;
+ pn_seconds_t timeout;
+ bool dynamic;
+ pn_data_t *properties;
+ pn_data_t *capabilities;
+ pn_data_t *outcomes;
+ pn_data_t *filter;
+};
+
struct pn_link_t {
pn_endpoint_t endpoint;
char *name;
pn_session_t *session;
- char *local_source;
- char *local_target;
- char *remote_source;
- char *remote_target;
+ pn_terminus_t source;
+ pn_terminus_t target;
+ pn_terminus_t remote_source;
+ pn_terminus_t remote_target;
pn_delivery_t *unsettled_head;
pn_delivery_t *unsettled_tail;
pn_delivery_t *current;
Modified: qpid/proton/trunk/proton-c/src/engine/engine.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/engine/engine.c?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/engine/engine.c (original)
+++ qpid/proton/trunk/proton-c/src/engine/engine.c Thu Oct 11 17:15:14 2012
@@ -350,14 +350,23 @@ void pn_link_close(pn_link_t *link)
if (link) pn_close((pn_endpoint_t *) link);
}
+void pn_terminus_free(pn_terminus_t *terminus)
+{
+ free(terminus->address);
+ pn_data_free(terminus->properties);
+ pn_data_free(terminus->capabilities);
+ pn_data_free(terminus->outcomes);
+ pn_data_free(terminus->filter);
+}
+
void pn_link_free(pn_link_t *link)
{
if (!link) return;
- free(link->local_source);
- free(link->local_target);
- free(link->remote_source);
- free(link->remote_target);
+ pn_terminus_free(&link->source);
+ pn_terminus_free(&link->target);
+ pn_terminus_free(&link->remote_source);
+ pn_terminus_free(&link->remote_target);
pn_remove_link(link->session, link);
while (link->settled_head) {
pn_delivery_t *d = link->settled_head;
@@ -821,15 +830,29 @@ pn_error_t *pn_transport_error(pn_transp
return transport->error;
}
+void pn_terminus_init(pn_terminus_t *terminus, pn_terminus_type_t type)
+{
+ terminus->type = type;
+ terminus->address = NULL;
+ terminus->durability = PN_NONDURABLE;
+ terminus->expiry_policy = PN_SESSION_CLOSE;
+ terminus->timeout = 0;
+ terminus->dynamic = false;
+ terminus->properties = pn_data(16);
+ terminus->capabilities = pn_data(16);
+ terminus->outcomes = pn_data(16);
+ terminus->filter = pn_data(16);
+}
+
void pn_link_init(pn_link_t *link, int type, pn_session_t *session, const char *name)
{
pn_endpoint_init(&link->endpoint, type, session->connection);
pn_add_link(session, link);
link->name = pn_strdup(name);
- link->local_source = NULL;
- link->local_target = NULL;
- link->remote_source = NULL;
- link->remote_target = NULL;
+ pn_terminus_init(&link->source, PN_SOURCE);
+ pn_terminus_init(&link->target, PN_TARGET);
+ pn_terminus_init(&link->remote_source, PN_UNSPECIFIED);
+ pn_terminus_init(&link->remote_target, PN_UNSPECIFIED);
link->settled_head = link->settled_tail = NULL;
link->unsettled_head = link->unsettled_tail = link->current = NULL;
link->unsettled_count = 0;
@@ -841,38 +864,154 @@ void pn_link_init(pn_link_t *link, int t
link->context = 0;
}
-const char *pn_link_get_source(pn_link_t *link)
+pn_terminus_t *pn_link_source(pn_link_t *link)
{
- return link ? link->local_source : NULL;
+ return link ? &link->source : NULL;
}
-const char *pn_link_get_target(pn_link_t *link)
+pn_terminus_t *pn_link_target(pn_link_t *link)
{
- return link ? link->local_target : NULL;
+ return link ? &link->target : NULL;
}
-void pn_link_set_source(pn_link_t *link, const char *source)
+pn_terminus_t *pn_link_remote_source(pn_link_t *link)
{
- if (!link) return;
- if (link->local_source) free(link->local_source);
- link->local_source = pn_strdup(source);
+ return link ? &link->remote_source : NULL;
}
-void pn_link_set_target(pn_link_t *link, const char *target)
+pn_terminus_t *pn_link_remote_target(pn_link_t *link)
{
- if (!link) return;
- if (link->local_target) free(link->local_target);
- link->local_target = pn_strdup(target);
+ return link ? &link->remote_target : NULL;
+}
+
+int pn_terminus_set_type(pn_terminus_t *terminus, pn_terminus_type_t type)
+{
+ if (!terminus) return PN_ARG_ERR;
+ terminus->type = type;
+ return 0;
+}
+
+pn_terminus_type_t pn_terminus_get_type(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->type : 0;
}
-const char *pn_link_remote_source(pn_link_t *link)
+const char *pn_terminus_get_address(pn_terminus_t *terminus)
{
- return link ? link->remote_source : NULL;
+ return terminus ? terminus->address : NULL;
}
-const char *pn_link_remote_target(pn_link_t *link)
+int pn_terminus_set_address(pn_terminus_t *terminus, const char *address)
{
- return link ? link->remote_target : NULL;
+ if (!terminus) return PN_ARG_ERR;
+ if (terminus->address) free(terminus->address);
+ terminus->address = pn_strdup(address);
+ return 0;
+}
+
+char *pn_bytes_strdup(pn_bytes_t str)
+{
+ return pn_strndup(str.start, str.size);
+}
+
+int pn_terminus_set_address_bytes(pn_terminus_t *terminus, pn_bytes_t address)
+{
+ if (!terminus) return PN_ARG_ERR;
+ if (terminus->address) free(terminus->address);
+ terminus->address = pn_bytes_strdup(address);
+ return 0;
+}
+
+pn_durability_t pn_terminus_get_durability(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->durability : 0;
+}
+
+int pn_terminus_set_durability(pn_terminus_t *terminus, pn_durability_t durability)
+{
+ if (!terminus) return PN_ARG_ERR;
+ terminus->durability = durability;
+ return 0;
+}
+
+pn_expiry_policy_t pn_terminus_get_expiry_policy(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->expiry_policy : 0;
+}
+
+int pn_terminus_set_expiry_policy(pn_terminus_t *terminus, pn_expiry_policy_t expiry_policy)
+{
+ if (!terminus) return PN_ARG_ERR;
+ terminus->expiry_policy = expiry_policy;
+ return 0;
+}
+
+pn_seconds_t pn_terminus_get_timeout(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->timeout : 0;
+}
+
+int pn_terminus_set_timeout(pn_terminus_t *terminus, pn_seconds_t timeout)
+{
+ if (!terminus) return PN_ARG_ERR;
+ terminus->timeout = timeout;
+ return 0;
+}
+
+bool pn_terminus_is_dynamic(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->dynamic : false;
+}
+
+int pn_terminus_set_dynamic(pn_terminus_t *terminus, bool dynamic)
+{
+ if (!terminus) return PN_ARG_ERR;
+ terminus->dynamic = dynamic;
+ return 0;
+}
+
+pn_data_t *pn_terminus_properties(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->properties : NULL;
+}
+
+pn_data_t *pn_terminus_capabilities(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->capabilities : NULL;
+}
+
+pn_data_t *pn_terminus_outcomes(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->outcomes : NULL;
+}
+
+pn_data_t *pn_terminus_filter(pn_terminus_t *terminus)
+{
+ return terminus ? terminus->filter : NULL;
+}
+
+int pn_terminus_copy(pn_terminus_t *terminus, pn_terminus_t *src)
+{
+ if (!terminus || !src) {
+ return PN_ARG_ERR;
+ }
+
+ terminus->type = src->type;
+ int err = pn_terminus_set_address(terminus, pn_terminus_get_address(src));
+ if (err) return err;
+ terminus->durability = src->durability;
+ terminus->expiry_policy = src->expiry_policy;
+ terminus->timeout = src->timeout;
+ terminus->dynamic = src->dynamic;
+ err = pn_data_copy(terminus->properties, src->properties);
+ if (err) return err;
+ err = pn_data_copy(terminus->capabilities, src->capabilities);
+ if (err) return err;
+ err = pn_data_copy(terminus->outcomes, src->outcomes);
+ if (err) return err;
+ err = pn_data_copy(terminus->filter, src->filter);
+ if (err) return err;
+ return 0;
}
pn_link_state_t *pn_link_get_state(pn_session_state_t *ssn_state, pn_link_t *link)
@@ -1163,11 +1302,6 @@ int pn_do_error(pn_transport_t *transpor
return PN_ERR;
}
-char *pn_bytes_strdup(pn_bytes_t str)
-{
- return pn_strndup(str.start, str.size);
-}
-
int pn_do_open(pn_dispatcher_t *disp)
{
pn_transport_t *transport = disp->context;
@@ -1236,17 +1370,40 @@ pn_link_state_t *pn_find_link(pn_session
return NULL;
}
+static pn_expiry_policy_t symbol2policy(pn_bytes_t symbol)
+{
+ if (!symbol.start)
+ return PN_SESSION_CLOSE;
+
+ if (!strncmp(symbol.start, "link-detach", symbol.size))
+ return PN_LINK_CLOSE;
+ if (!strncmp(symbol.start, "session-end", symbol.size))
+ return PN_SESSION_CLOSE;
+ if (!strncmp(symbol.start, "connection-close", symbol.size))
+ return PN_CONNECTION_CLOSE;
+ if (!strncmp(symbol.start, "never", symbol.size))
+ return PN_NEVER;
+
+ return PN_SESSION_CLOSE;
+}
+
int pn_do_attach(pn_dispatcher_t *disp)
{
pn_transport_t *transport = disp->context;
pn_bytes_t name;
uint32_t handle;
bool is_sender;
- pn_bytes_t source;
- pn_bytes_t target;
+ pn_bytes_t source, target;
+ pn_durability_t src_dr, tgt_dr;
+ pn_bytes_t src_exp, tgt_exp;
+ pn_seconds_t src_timeout, tgt_timeout;
+ bool src_dynamic, tgt_dynamic;
pn_sequence_t idc;
- int err = pn_scan_args(disp, "D.[SIo..D.[S]D.[S]..I]", &name, &handle,
- &is_sender, &source, &target, &idc);
+ int err = pn_scan_args(disp, "D.[SIo..D.[SIsIo]D.[SIsIo]..I]", &name, &handle,
+ &is_sender,
+ &source, &src_dr, &src_exp, &src_timeout, &src_dynamic,
+ &target, &tgt_dr, &tgt_exp, &tgt_timeout, &tgt_dynamic,
+ &idc);
if (err) return err;
char strname[name.size + 1];
strncpy(strname, name.start, name.size);
@@ -1254,22 +1411,65 @@ int pn_do_attach(pn_dispatcher_t *disp)
pn_session_state_t *ssn_state = pn_channel_state(transport, disp->channel);
pn_link_state_t *link_state = pn_find_link(ssn_state, name, is_sender);
+ pn_link_t *link;
if (!link_state) {
- pn_link_t *link;
if (is_sender) {
link = (pn_link_t *) pn_sender(ssn_state->session, strname);
} else {
link = (pn_link_t *) pn_receiver(ssn_state->session, strname);
}
link_state = pn_link_get_state(ssn_state, link);
+ } else {
+ link = link_state->link;
}
pn_map_handle(ssn_state, handle, link_state);
- PN_SET_REMOTE(link_state->link->endpoint.state, PN_REMOTE_ACTIVE);
- if (source.start)
- link_state->link->remote_source = pn_bytes_strdup(source);
- if (target.start)
- link_state->link->remote_target = pn_bytes_strdup(target);
+ PN_SET_REMOTE(link->endpoint.state, PN_REMOTE_ACTIVE);
+ pn_terminus_t *rsrc = &link_state->link->remote_source;
+ if (source.start) {
+ pn_terminus_set_type(rsrc, PN_SOURCE);
+ pn_terminus_set_address_bytes(rsrc, source);
+ pn_terminus_set_durability(rsrc, src_dr);
+ pn_terminus_set_expiry_policy(rsrc, symbol2policy(src_exp));
+ pn_terminus_set_timeout(rsrc, src_timeout);
+ pn_terminus_set_dynamic(rsrc, src_dynamic);
+ } else {
+ pn_terminus_set_type(rsrc, PN_UNSPECIFIED);
+ }
+ pn_terminus_t *rtgt = &link_state->link->remote_target;
+ if (target.start) {
+ pn_terminus_set_type(rtgt, PN_TARGET);
+ pn_terminus_set_address_bytes(rtgt, target);
+ pn_terminus_set_durability(rtgt, tgt_dr);
+ pn_terminus_set_expiry_policy(rtgt, symbol2policy(tgt_exp));
+ pn_terminus_set_timeout(rtgt, tgt_timeout);
+ pn_terminus_set_dynamic(rtgt, tgt_dynamic);
+ } else {
+ pn_terminus_set_type(rtgt, PN_UNSPECIFIED);
+ }
+
+ pn_data_clear(link->remote_source.properties);
+ pn_data_clear(link->remote_source.filter);
+ pn_data_clear(link->remote_source.outcomes);
+ pn_data_clear(link->remote_source.capabilities);
+ pn_data_clear(link->remote_target.properties);
+ pn_data_clear(link->remote_target.capabilities);
+
+ err = pn_scan_args(disp, "D.[.....D.[.....C.C.CC]D.[.....CC]",
+ link->remote_source.properties,
+ link->remote_source.filter,
+ link->remote_source.outcomes,
+ link->remote_source.capabilities,
+ link->remote_target.properties,
+ link->remote_target.capabilities);
+ if (err) return err;
+
+ pn_data_rewind(link->remote_source.properties);
+ pn_data_rewind(link->remote_source.filter);
+ pn_data_rewind(link->remote_source.outcomes);
+ pn_data_rewind(link->remote_source.capabilities);
+ pn_data_rewind(link->remote_target.properties);
+ pn_data_rewind(link->remote_target.capabilities);
if (!is_sender) {
link_state->delivery_count = idc;
@@ -1286,9 +1486,11 @@ int pn_do_transfer(pn_dispatcher_t *disp
pn_transport_t *transport = disp->context;
uint32_t handle;
pn_bytes_t tag;
+ bool id_present;
pn_sequence_t id;
bool more;
- int err = pn_scan_args(disp, "D.[IIz..o]", &handle, &id, &tag, &more);
+ int err = pn_scan_args(disp, "D.[I?Iz..o]", &handle, &id_present, &id, &tag,
+ &more);
if (err) return err;
pn_session_state_t *ssn_state = pn_channel_state(transport, disp->channel);
pn_link_state_t *link_state = pn_handle_state(ssn_state, handle);
@@ -1311,7 +1513,7 @@ int pn_do_transfer(pn_dispatcher_t *disp
delivery = pn_delivery(link, pn_dtag(tag.start, tag.size));
pn_delivery_state_t *state = pn_delivery_buffer_push(incoming, delivery);
delivery->transport_context = state;
- if (id != state->id) {
+ if (id_present && id != state->id) {
int err = pn_do_error(transport, "amqp:session:invalid-field",
"sequencing error, expected delivery-id %u, got %u",
state->id, id);
@@ -1653,6 +1855,22 @@ int pn_process_ssn_setup(pn_transport_t
return 0;
}
+static const char *expiry_symbol(pn_expiry_policy_t policy)
+{
+ switch (policy)
+ {
+ case PN_LINK_CLOSE:
+ return "link-detach";
+ case PN_SESSION_CLOSE:
+ return NULL;
+ case PN_CONNECTION_CLOSE:
+ return "connection-close";
+ case PN_NEVER:
+ return "never";
+ }
+ return NULL;
+}
+
int pn_process_link_setup(pn_transport_t *transport, pn_endpoint_t *endpoint)
{
if (transport->open_sent && (endpoint->type == SENDER ||
@@ -1667,12 +1885,28 @@ int pn_process_link_setup(pn_transport_t
// XXX
state->local_handle = link->id;
int err = pn_post_frame(transport->disp, ssn_state->local_channel,
- "DL[SIonn?DL[S]?DL[S]nnI]", ATTACH,
+ "DL[SIonn?DL[SIsIoCnCnCC]?DL[SIsIoCC]nnI]", ATTACH,
link->name,
state->local_handle,
endpoint->type == RECEIVER,
- (bool) link->local_source, SOURCE, link->local_source,
- (bool) link->local_target, TARGET, link->local_target,
+ (bool) link->source.type, SOURCE,
+ link->source.address,
+ link->source.durability,
+ expiry_symbol(link->source.expiry_policy),
+ link->source.timeout,
+ link->source.dynamic,
+ link->source.properties,
+ link->source.filter,
+ link->source.outcomes,
+ link->source.capabilities,
+ (bool) link->target.type, TARGET,
+ link->target.address,
+ link->target.durability,
+ expiry_symbol(link->target.expiry_policy),
+ link->target.timeout,
+ link->target.dynamic,
+ link->target.properties,
+ link->target.capabilities,
0);
if (err) return err;
}
Modified: qpid/proton/trunk/proton-c/src/message/message.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/message/message.c?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/message/message.c (original)
+++ qpid/proton/trunk/proton-c/src/message/message.c Thu Oct 11 17:15:14 2012
@@ -67,6 +67,9 @@ struct pn_message_t {
pn_buffer_t *reply_to_group_id;
pn_data_t *data;
+ pn_data_t *instructions;
+ pn_data_t *annotations;
+ pn_data_t *properties;
pn_data_t *body;
pn_format_t format;
@@ -95,8 +98,11 @@ pn_message_t *pn_message()
msg->group_id = NULL;
msg->group_sequence = 0;
msg->reply_to_group_id = NULL;
- msg->data = NULL;
- msg->body = NULL;
+ msg->data = pn_data(16);
+ msg->instructions = pn_data(16);
+ msg->annotations = pn_data(16);
+ msg->properties = pn_data(16);
+ msg->body = pn_data(16);
msg->format = PN_DATA;
msg->parser = NULL;
msg->error = pn_error();
@@ -115,6 +121,9 @@ void pn_message_free(pn_message_t *msg)
pn_buffer_free(msg->group_id);
pn_buffer_free(msg->reply_to_group_id);
pn_data_free(msg->data);
+ pn_data_free(msg->instructions);
+ pn_data_free(msg->annotations);
+ pn_data_free(msg->properties);
pn_data_free(msg->body);
pn_parser_free(msg->parser);
pn_error_free(msg->error);
@@ -142,8 +151,11 @@ void pn_message_clear(pn_message_t *msg)
if (msg->group_id) pn_buffer_clear(msg->group_id);
msg->group_sequence = 0;
if (msg->reply_to_group_id) pn_buffer_clear(msg->reply_to_group_id);
- if (msg->data) pn_data_clear(msg->data);
- if (msg->body) pn_data_clear(msg->body);
+ pn_data_clear(msg->data);
+ pn_data_clear(msg->instructions);
+ pn_data_clear(msg->annotations);
+ pn_data_clear(msg->properties);
+ pn_data_clear(msg->body);
}
int pn_message_errno(pn_message_t *msg)
@@ -414,14 +426,7 @@ int pn_message_decode(pn_message_t *msg,
{
if (!msg || !bytes || !size) return PN_ARG_ERR;
- if (!msg->data) {
- msg->data = pn_data(64);
- }
- if (!msg->body) {
- msg->body = pn_data(64);
- }
-
- pn_data_clear(msg->body);
+ pn_message_clear(msg);
while (size) {
pn_data_clear(msg->data);
@@ -435,10 +440,15 @@ int pn_message_decode(pn_message_t *msg,
int err = pn_data_scan(msg->data, "D?L.", &scanned, &desc);
if (err) return pn_error_format(msg->error, err, "data error: %s",
pn_data_error(msg->data));
- if (!scanned){
+ if (!scanned) {
desc = 0;
}
+ pn_data_rewind(msg->data);
+ pn_data_next(msg->data);
+ pn_data_enter(msg->data);
+ pn_data_next(msg->data);
+
switch (desc) {
case HEADER:
pn_data_scan(msg->data, "D.[oBIoI]", &msg->durable, &msg->priority,
@@ -475,14 +485,23 @@ int pn_message_decode(pn_message_t *msg,
}
break;
case DELIVERY_ANNOTATIONS:
+ pn_data_narrow(msg->data);
+ err = pn_data_copy(msg->instructions, msg->data);
+ if (err) return err;
+ break;
case MESSAGE_ANNOTATIONS:
+ pn_data_narrow(msg->data);
+ err = pn_data_copy(msg->annotations, msg->data);
+ if (err) return err;
+ break;
+ case APPLICATION_PROPERTIES:
+ pn_data_narrow(msg->data);
+ err = pn_data_copy(msg->properties, msg->data);
+ if (err) return err;
break;
default:
- {
- pn_data_t *data = msg->body;
- msg->body = msg->data;
- msg->data = data;
- }
+ err = pn_data_copy(msg->body, msg->data);
+ if (err) return err;
break;
}
}
@@ -510,6 +529,30 @@ int pn_message_encode(pn_message_t *msg,
return pn_error_format(msg->error, err, "data error: %s",
pn_data_error(msg->data));
+ if (pn_data_size(msg->instructions)) {
+ pn_data_put_described(msg->data);
+ pn_data_enter(msg->data);
+ pn_data_put_ulong(msg->data, DELIVERY_ANNOTATIONS);
+ pn_data_rewind(msg->instructions);
+ err = pn_data_append(msg->data, msg->instructions);
+ if (err)
+ return pn_error_format(msg->error, err, "data error: %s",
+ pn_data_error(msg->data));
+ pn_data_exit(msg->data);
+ }
+
+ if (pn_data_size(msg->annotations)) {
+ pn_data_put_described(msg->data);
+ pn_data_enter(msg->data);
+ pn_data_put_ulong(msg->data, MESSAGE_ANNOTATIONS);
+ pn_data_rewind(msg->annotations);
+ err = pn_data_append(msg->data, msg->annotations);
+ if (err)
+ return pn_error_format(msg->error, err, "data error: %s",
+ pn_data_error(msg->data));
+ pn_data_exit(msg->data);
+ }
+
err = pn_data_fill(msg->data, "DL[nzSSSnssttSIS]", PROPERTIES,
pn_buffer_bytes(msg->user_id),
pn_buffer_str(msg->address),
@@ -526,24 +569,39 @@ int pn_message_encode(pn_message_t *msg,
return pn_error_format(msg->error, err, "data error: %s",
pn_data_error(msg->data));
+ if (pn_data_size(msg->properties)) {
+ pn_data_put_described(msg->data);
+ pn_data_enter(msg->data);
+ pn_data_put_ulong(msg->data, APPLICATION_PROPERTIES);
+ pn_data_rewind(msg->properties);
+ err = pn_data_append(msg->data, msg->properties);
+ if (err)
+ return pn_error_format(msg->error, err, "data error: %s",
+ pn_data_error(msg->data));
+ pn_data_exit(msg->data);
+ }
+
size_t remaining = *size;
- ssize_t encoded = pn_data_encode(msg->data, bytes, remaining);
- if (encoded < 0)
- return pn_error_format(msg->error, encoded, "data error: %s",
- pn_data_error(msg->data));
+ ssize_t encoded;
- bytes += encoded;
- remaining -= encoded;
+ pn_data_t *sections[2] = {
+ msg->data, msg->body
+ };
+
+ for (int i = 0; i < 2; i++) {
+ encoded = pn_data_encode(sections[i], bytes, remaining);
+ if (encoded < 0)
+ return pn_error_format(msg->error, encoded, "data error: %s",
+ pn_data_error(sections[i]));
- encoded = pn_data_encode(msg->body, bytes, remaining);
- if (encoded < 0)
- return pn_error_format(msg->error, encoded, "data error: %s",
- pn_data_error(msg->body));
- bytes += encoded;
- remaining -= encoded;
+ bytes += encoded;
+ remaining -= encoded;
+ }
*size -= remaining;
+ pn_data_clear(msg->data);
+
return 0;
}
@@ -683,7 +741,7 @@ int pn_message_save_text(pn_message_t *m
{
if (!msg) return PN_ARG_ERR;
- if (!msg->body) {
+ if (pn_data_size(msg->body) == 0) {
*size = 0;
return 0;
}
@@ -732,3 +790,23 @@ int pn_message_save_json(pn_message_t *m
return PN_ERR;
}
+
+pn_data_t *pn_message_instructions(pn_message_t *msg)
+{
+ return msg ? msg->instructions : NULL;
+}
+
+pn_data_t *pn_message_annotations(pn_message_t *msg)
+{
+ return msg ? msg->annotations : NULL;
+}
+
+pn_data_t *pn_message_properties(pn_message_t *msg)
+{
+ return msg ? msg->properties : NULL;
+}
+
+pn_data_t *pn_message_body(pn_message_t *msg)
+{
+ return msg ? msg->body : NULL;
+}
Modified: qpid/proton/trunk/proton-c/src/messenger.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/messenger.c?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/messenger.c (original)
+++ qpid/proton/trunk/proton-c/src/messenger.c Thu Oct 11 17:15:14 2012
@@ -213,8 +213,8 @@ void pn_messenger_endpoints(pn_messenger
pn_link_t *link = pn_link_head(conn, PN_LOCAL_UNINIT);
while (link) {
- pn_link_set_source(link, pn_link_remote_source(link));
- pn_link_set_target(link, pn_link_remote_target(link));
+ pn_terminus_copy(pn_link_source(link), pn_link_remote_source(link));
+ pn_terminus_copy(pn_link_target(link), pn_link_remote_target(link));
pn_link_open(link);
link = pn_link_next(link, PN_LOCAL_UNINIT);
}
@@ -286,7 +286,6 @@ int pn_messenger_tsync(pn_messenger_t *m
messenger->private_key,
messenger->password);
}
- pn_ssl_set_peer_authentication(ssl, PN_SSL_NO_VERIFY_PEER, NULL);
if (!(scheme && !strcmp(scheme, "amqps"))) {
pn_ssl_allow_unsecured_client(ssl);
}
@@ -429,7 +428,7 @@ pn_connection_t *pn_messenger_resolve(pn
pn_ssl_set_trusted_ca_db(ssl, messenger->trusted_certificates);
pn_ssl_set_peer_authentication(ssl, PN_SSL_VERIFY_PEER, NULL);
} else {
- pn_ssl_set_peer_authentication(ssl, PN_SSL_NO_VERIFY_PEER, NULL);
+ pn_ssl_set_peer_authentication(ssl, PN_SSL_ANONYMOUS_PEER, NULL);
}
}
@@ -465,7 +464,8 @@ pn_link_t *pn_messenger_link(pn_messenge
while (link) {
if (pn_link_is_sender(link) == sender) {
const char *terminus = pn_link_is_sender(link) ?
- pn_link_get_target(link) : pn_link_get_source(link);
+ pn_terminus_get_address(pn_link_target(link)) :
+ pn_terminus_get_address(pn_link_source(link));
if (pn_streq(name, terminus))
return link;
}
@@ -477,11 +477,11 @@ pn_link_t *pn_messenger_link(pn_messenge
link = sender ? pn_sender(ssn, "sender-xxx") : pn_receiver(ssn, "receiver-xxx");
// XXX
if (sender) {
- pn_link_set_target(link, name);
- pn_link_set_source(link, name);
+ pn_terminus_set_address(pn_link_target(link), name);
+ pn_terminus_set_address(pn_link_source(link), name);
} else {
- pn_link_set_target(link, name);
- pn_link_set_source(link, name);
+ pn_terminus_set_address(pn_link_target(link), name);
+ pn_terminus_set_address(pn_link_source(link), name);
}
pn_link_open(link);
return link;
@@ -683,6 +683,7 @@ int pn_messenger_get(pn_messenger_t *mes
return pn_error_format(messenger->error, n, "didn't receive pending bytes: %zi", n);
}
n = pn_link_recv(l, encoded + pending, 1);
+ pn_delivery_update(d, PN_ACCEPTED);
pn_delivery_settle(d);
if (n != PN_EOS) {
return pn_error_format(messenger->error, n, "PN_EOS expected");
Modified: qpid/proton/trunk/proton-c/src/proton.c
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-c/src/proton.c?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-c/src/proton.c (original)
+++ qpid/proton/trunk/proton-c/src/proton.c Thu Oct 11 17:15:14 2012
@@ -147,10 +147,10 @@ void server_callback(pn_connector_t *cto
pn_link_t *link = pn_link_head(conn, PN_LOCAL_UNINIT | PN_REMOTE_ACTIVE);
while (link) {
- printf("%s, %s\n", pn_link_remote_source(link),
- pn_link_remote_target(link));
- pn_link_set_source(link, pn_link_remote_source(link));
- pn_link_set_target(link, pn_link_remote_target(link));
+ printf("%s, %s\n", pn_terminus_get_address(pn_link_remote_source(link)),
+ pn_terminus_get_address(pn_link_remote_target(link)));
+ pn_terminus_copy(pn_link_source(link), pn_link_remote_source(link));
+ pn_terminus_copy(pn_link_target(link), pn_link_remote_target(link));
pn_link_open(link);
if (pn_link_is_receiver(link)) {
pn_link_flow(link, 100);
@@ -297,7 +297,7 @@ void client_callback(pn_connector_t *cto
if (ctx->send_count) {
pn_link_t *snd = pn_sender(ssn, "sender");
- pn_link_set_target(snd, ctx->address);
+ pn_terminus_set_address(pn_link_target(snd), ctx->address);
pn_link_open(snd);
char buf[16];
@@ -309,7 +309,7 @@ void client_callback(pn_connector_t *cto
if (ctx->recv_count) {
pn_link_t *rcv = pn_receiver(ssn, "receiver");
- pn_link_set_source(rcv, ctx->address);
+ pn_terminus_set_address(pn_link_source(rcv), ctx->address);
pn_link_open(rcv);
pn_link_flow(rcv, ctx->recv_count < ctx->high ? ctx->recv_count : ctx->high);
}
Modified: qpid/proton/trunk/proton-j/src/main/scripts/proton.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/proton-j/src/main/scripts/proton.py?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/proton-j/src/main/scripts/proton.py (original)
+++ qpid/proton/trunk/proton-j/src/main/scripts/proton.py Thu Oct 11 17:15:14 2012
@@ -174,24 +174,20 @@ class Link(Endpoint):
def __init__(self, impl):
self.impl = impl
- def _set_source(self, source):
- self.impl.setLocalSourceAddress(source)
- def _get_source(self):
- return self.impl.getLocalSourceAddress()
- source = property(_get_source, _set_source)
-
- def _set_target(self, target):
- self.impl.setLocalTargetAddress(target)
- def _get_target(self):
- return self.impl.getLocalTargetAddress()
- target = property(_get_target, _set_target)
+ @property
+ def source(self):
+ return Terminus(self.impl, "LocalSource")
+
+ @property
+ def target(self):
+ return Terminus(self.impl, "LocalTarget")
@property
def remote_source(self):
- return self.impl.getRemoteSourceAddress()
+ return Terminus(self.impl, "RemoteSource")
@property
def remote_target(self):
- return self.impl.getRemoteTargetAddress()
+ return Terminus(self.impl, "RemoteTarget")
@property
def session(self):
@@ -226,6 +222,40 @@ class Link(Endpoint):
def next(self, mask):
return self.impl.next(*self._enums(mask))
+class DataDummy:
+
+ def format(self):
+ pass
+
+ def put_array(self, *args, **kwargs):
+ raise Skipped()
+
+class Terminus(object):
+
+ UNSPECIFIED = None
+
+ def __init__(self, impl, prefix):
+ self.impl = impl
+ self.prefix = prefix
+ self.type = None
+ self.timeout = None
+ self.durability = None
+ self.expiry_policy = None
+ self.dynamic = None
+ self.properties = DataDummy()
+ self.outcomes = DataDummy()
+ self.filter = DataDummy()
+ self.capabilities = DataDummy()
+
+ def _get_address(self):
+ return getattr(self.impl, "get%sAddress" % self.prefix)()
+ def _set_address(self, address):
+ getattr(self.impl, "set%sAddress" % self.prefix)(address)
+ address = property(_get_address, _set_address)
+
+ def copy(self, src):
+ self.address = src.address
+
class Sender(Link):
def offered(self, n):
@@ -349,6 +379,8 @@ class Transport(object):
class Data(object):
+ SYMBOL = None
+
def __init__(self, *args, **kwargs):
raise Skipped()
@@ -514,5 +546,6 @@ class SSL(object):
__all__ = ["Messenger", "Message", "ProtonException", "MessengerException",
"MessageException", "Timeout", "Data", "Endpoint", "Connection",
- "Session", "Link", "Sender", "Receiver", "Delivery", "Transport",
- "TransportException", "SASL", "SSL", "PN_SESSION_WINDOW"]
+ "Session", "Link", "Terminus", "Sender", "Receiver", "Delivery",
+ "Transport", "TransportException", "SASL", "SSL",
+ "PN_SESSION_WINDOW"]
Modified: qpid/proton/trunk/tests/proton_tests/codec.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/codec.py?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/tests/proton_tests/codec.py (original)
+++ qpid/proton/trunk/tests/proton_tests/codec.py Thu Oct 11 17:15:14 2012
@@ -247,3 +247,17 @@ class DataTest(Test):
def testDecimal128(self):
self._test("decimal128", "fdsaasdf;lkjjkl;", "x"*16 )
+
+ def testCopy(self):
+ self.data.put_described()
+ self.data.enter()
+ self.data.put_ulong(123)
+ self.data.put_map()
+ self.data.enter()
+ self.data.put_string("pi")
+ self.data.put_double(3.14159265359)
+
+ dst = Data()
+ dst.copy(self.data)
+
+ assert dst.format() == self.data.format()
Modified: qpid/proton/trunk/tests/proton_tests/engine.py
URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/engine.py?rev=1397174&r1=1397173&r2=1397174&view=diff
==============================================================================
--- qpid/proton/trunk/tests/proton_tests/engine.py (original)
+++ qpid/proton/trunk/tests/proton_tests/engine.py Thu Oct 11 17:15:14 2012
@@ -352,28 +352,85 @@ class LinkTest(Test):
self.snd.close()
self.pump()
- def _test_source_target(self, source, target):
- self.snd.source = source
- self.snd.target = target
+ def assertEqualTermini(self, t1, t2):
+ assert t1.type == t2.type, (t1.type, t2.type)
+ assert t1.address == t2.address, (t1.address, t2.address)
+ assert t1.durability == t2.durability, (t1.durability, t2.durability)
+ assert t1.expiry_policy == t2.expiry_policy, (t1.expiry_policy, t2.expiry_policy)
+ assert t1.timeout == t2.timeout, (t1.timeout, t2.timeout)
+ assert t1.dynamic == t2.dynamic, (t1.dynamic, t2.dynamic)
+ for attr in ["properties", "capabilities", "outcomes", "filter"]:
+ d1 = getattr(t1, attr)
+ d2 = getattr(t2, attr)
+ assert d1.format() == d2.format(), (attr, d1.format(), d2.format())
+
+ def _test_source_target(self, config_source, config_target):
+ if config_source is None:
+ self.snd.source.type = Terminus.UNSPECIFIED
+ else:
+ config_source(self.snd.source)
+ if config_target is None:
+ self.snd.target.type = Terminus.UNSPECIFIED
+ else:
+ config_target(self.snd.target)
self.snd.open()
self.pump()
- assert self.rcv.remote_source == self.snd.source
- assert self.rcv.remote_target == self.snd.target
- self.rcv.target = self.rcv.remote_target
- self.rcv.source = self.rcv.remote_source
+ self.assertEqualTermini(self.rcv.remote_source, self.snd.source)
+ self.assertEqualTermini(self.rcv.remote_target, self.snd.target)
+ self.rcv.target.copy(self.rcv.remote_target)
+ self.rcv.source.copy(self.rcv.remote_source)
self.rcv.open()
self.pump()
- assert self.snd.remote_target == self.snd.target
- assert self.snd.remote_source == self.snd.source
+ self.assertEqualTermini(self.snd.remote_target, self.snd.target)
+ self.assertEqualTermini(self.snd.remote_source, self.snd.source)
def test_source_target(self):
- self._test_source_target("source", "target")
+ self._test_source_target(TerminusConfig(address="source"),
+ TerminusConfig(address="target"))
def test_source(self):
- self._test_source_target("source", None)
+ self._test_source_target(TerminusConfig(address="source"), None)
def test_target(self):
- self._test_source_target(None, "target")
+ self._test_source_target(None, TerminusConfig(address="target"))
+
+ def test_source_target_full(self):
+ self._test_source_target(TerminusConfig(address="source",
+ timeout=3,
+ filter=[("int", 1), ("symbol", "two"), ("string", "three")],
+ capabilities=["one", "two", "three"]),
+ TerminusConfig(address="source",
+ timeout=7,
+ capabilities=[]))
+
+class TerminusConfig:
+
+ def __init__(self, address=None, timeout=None, durability=None, filter=None,
+ capabilities=None):
+ self.address = address
+ self.timeout = timeout
+ self.durability = durability
+ self.filter = filter
+ self.capabilities = capabilities
+
+ def __call__(self, terminus):
+ if self.address is not None:
+ terminus.address = self.address
+ if self.timeout is not None:
+ terminus.timeout = self.timeout
+ if self.durability is not None:
+ terminus.durability = self.durability
+ if self.capabilities is not None:
+ terminus.capabilities.put_array(False, Data.SYMBOL)
+ terminus.capabilities.enter()
+ for c in self.capabilities:
+ terminus.capabilities.put_symbol(c)
+ if self.filter is not None:
+ terminus.filter.put_list()
+ terminus.filter.enter()
+ for (t, v) in self.filter:
+ setter = getattr(terminus.filter, "put_%s" % t)
+ setter(v)
class TransferTest(Test):
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org