You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2013/02/18 11:18:29 UTC
svn commit: r1447183 [5/7] - in /qpid/branches/java-broker-config-qpid-4390:
./ qpid/ qpid/bin/ qpid/cpp/bindings/ qpid/cpp/bindings/qpid/dotnet/
qpid/cpp/bindings/qpid/examples/perl/ qpid/cpp/bindings/qpid/perl/
qpid/cpp/bindings/qpid/perl/lib/ qpid/c...
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/ctools.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/ctools.h?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/ctools.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/ctools.h Mon Feb 18 10:18:24 2013
@@ -37,11 +37,11 @@
#define DEQ_LINKS(t) t *prev; t *next
-#define DEQ_INIT(d) do { d.head = 0; d.tail = 0; d.scratch = 0; d.size = 0; } while (0)
+#define DEQ_INIT(d) do { (d).head = 0; (d).tail = 0; (d).scratch = 0; (d).size = 0; } while (0)
#define DEQ_ITEM_INIT(i) do { (i)->next = 0; (i)->prev = 0; } while(0)
-#define DEQ_HEAD(d) (d.head)
-#define DEQ_TAIL(d) (d.tail)
-#define DEQ_SIZE(d) (d.size)
+#define DEQ_HEAD(d) ((d).head)
+#define DEQ_TAIL(d) ((d).tail)
+#define DEQ_SIZE(d) ((d).size)
#define DEQ_NEXT(i) (i)->next
#define DEQ_PREV(i) (i)->prev
@@ -49,67 +49,67 @@
do { \
CT_ASSERT((i)->next == 0); \
CT_ASSERT((i)->prev == 0); \
- if (d.head) { \
- (i)->next = d.head; \
- d.head->prev = i; \
+ if ((d).head) { \
+ (i)->next = (d).head; \
+ (d).head->prev = i; \
} else { \
- d.tail = i; \
+ (d).tail = i; \
(i)->next = 0; \
- CT_ASSERT(d.size == 0); \
+ CT_ASSERT((d).size == 0); \
} \
(i)->prev = 0; \
- d.head = i; \
- d.size++; \
+ (d).head = i; \
+ (d).size++; \
} while (0)
#define DEQ_INSERT_TAIL(d,i) \
do { \
CT_ASSERT((i)->next == 0); \
CT_ASSERT((i)->prev == 0); \
- if (d.tail) { \
- (i)->prev = d.tail; \
- d.tail->next = i; \
+ if ((d).tail) { \
+ (i)->prev = (d).tail; \
+ (d).tail->next = i; \
} else { \
- d.head = i; \
+ (d).head = i; \
(i)->prev = 0; \
- CT_ASSERT(d.size == 0); \
+ CT_ASSERT((d).size == 0); \
} \
(i)->next = 0; \
- d.tail = i; \
- d.size++; \
+ (d).tail = i; \
+ (d).size++; \
} while (0)
-#define DEQ_REMOVE_HEAD(d) \
-do { \
- CT_ASSERT(d.head); \
- if (d.head) { \
- d.scratch = d.head; \
- d.head = d.head->next; \
- if (d.head == 0) { \
- d.tail = 0; \
- CT_ASSERT(d.size == 1); \
+#define DEQ_REMOVE_HEAD(d) \
+do { \
+ CT_ASSERT((d).head); \
+ if ((d).head) { \
+ (d).scratch = (d).head; \
+ (d).head = (d).head->next; \
+ if ((d).head == 0) { \
+ (d).tail = 0; \
+ CT_ASSERT((d).size == 1); \
} else \
- d.head->prev = 0; \
- d.size--; \
- d.scratch->next = 0; \
- d.scratch->prev = 0; \
+ (d).head->prev = 0; \
+ (d).size--; \
+ (d).scratch->next = 0; \
+ (d).scratch->prev = 0; \
} \
} while (0)
#define DEQ_REMOVE_TAIL(d) \
do { \
- CT_ASSERT(d.tail); \
- if (d.tail) { \
- d.scratch = d.tail; \
- d.tail = d.tail->prev; \
- if (d.tail == 0) { \
- d.head = 0; \
- CT_ASSERT(d.size == 1); \
+ CT_ASSERT((d).tail); \
+ if ((d).tail) { \
+ (d).scratch = (d).tail; \
+ (d).tail = (d).tail->prev; \
+ if ((d).tail == 0) { \
+ (d).head = 0; \
+ CT_ASSERT((d).size == 1); \
} else \
- d.tail->next = 0; \
- d.size--; \
- d.scratch->next = 0; \
- d.scratch->prev = 0; \
+ (d).tail->next = 0; \
+ (d).size--; \
+ (d).scratch->next = 0; \
+ (d).scratch->prev = 0; \
} \
} while (0)
@@ -120,11 +120,11 @@ do { \
if ((a)->next) \
(a)->next->prev = (i); \
else \
- d.tail = (i); \
+ (d).tail = (i); \
(i)->next = (a)->next; \
(i)->prev = (a); \
(a)->next = (i); \
- d.size++; \
+ (d).size++; \
} while (0)
#define DEQ_REMOVE(d,i) \
@@ -132,15 +132,15 @@ do {
if ((i)->next) \
(i)->next->prev = (i)->prev; \
else \
- d.tail = (i)->prev; \
+ (d).tail = (i)->prev; \
if ((i)->prev) \
(i)->prev->next = (i)->next; \
else \
- d.head = (i)->next; \
- d.size--; \
+ (d).head = (i)->next; \
+ (d).size--; \
(i)->next = 0; \
(i)->prev = 0; \
- CT_ASSERT(d.size || (!d.head && !d.tail)); \
+ CT_ASSERT((d).size || (!(d).head && !(d).tail)); \
} while (0)
#endif
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/iterator.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/iterator.h?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/iterator.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/iterator.h Mon Feb 18 10:18:24 2013
@@ -19,8 +19,7 @@
* under the License.
*/
-
-typedef struct nx_buffer_t nx_buffer_t;
+#include <qpid/nexus/buffer.h>
/**
* The field iterator is used to access fields within a buffer chain.
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/message.h
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/message.h?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/message.h (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/include/qpid/nexus/message.h Mon Feb 18 10:18:24 2013
@@ -21,99 +21,107 @@
#include <proton/engine.h>
#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/alloc.h>
#include <qpid/nexus/iterator.h>
+#include <qpid/nexus/buffer.h>
+#include <qpid/nexus/iovec.h>
+
+// Callback for status change (confirmed persistent, loaded-in-memory, etc.)
typedef struct nx_message_t nx_message_t;
-typedef struct nx_buffer_t nx_buffer_t;
-DEQ_DECLARE(nx_buffer_t, nx_buffer_list_t);
DEQ_DECLARE(nx_message_t, nx_message_list_t);
-typedef struct {
- nx_buffer_t *buffer; // Buffer that contains the first octet of the field, null if the field is not present
- size_t offset; // Offset in the buffer to the first octet
- size_t length; // Length of the field or zero if unneeded
- int parsed; // non-zero iff the buffer chain has been parsed to find this field
-} nx_field_location_t;
-
-
-// TODO - consider using pointers to nx_field_location_t below to save memory
struct nx_message_t {
DEQ_LINKS(nx_message_t);
- nx_buffer_list_t buffers; // The buffer chain containing the message
- pn_delivery_t *in_delivery; // The delivery on which the message arrived
- pn_delivery_t *out_delivery; // The delivery on which the message was last sent
- nx_field_location_t section_message_header; // The message header list
- nx_field_location_t section_delivery_annotation; // The delivery annotation map
- nx_field_location_t section_message_annotation; // The message annotation map
- nx_field_location_t section_message_properties; // The message properties list
- nx_field_location_t section_application_properties; // The application properties list
- nx_field_location_t section_body; // The message body: Data
- nx_field_location_t section_footer; // The footer
- nx_field_location_t field_user_id; // The string value of the user-id
- nx_field_location_t field_to; // The string value of the to field
- nx_field_location_t body; // The body of the message
- nx_field_location_t compose_length;
- nx_field_location_t compose_count;
- uint32_t length;
- uint32_t count;
+ // Private members not listed here.
};
-struct nx_buffer_t {
- DEQ_LINKS(nx_buffer_t);
- unsigned int size;
-};
-
-typedef struct {
- size_t buffer_size;
- unsigned long buffer_preallocation_count;
- unsigned long buffer_rebalancing_batch_count;
- unsigned long buffer_local_storage_max;
- unsigned long buffer_free_list_max;
- unsigned long message_allocation_batch_count;
- unsigned long message_rebalancing_batch_count;
- unsigned long message_local_storage_max;
-} nx_allocator_config_t;
-
-const nx_allocator_config_t *nx_allocator_default_config(void);
-
-void nx_allocator_initialize(const nx_allocator_config_t *config);
-void nx_allocator_finalize(void);
-
-//
-// Functions for per-thread allocators.
-//
-nx_message_t *nx_allocate_message(void);
-nx_buffer_t *nx_allocate_buffer(void);
-void nx_free_message(nx_message_t *msg);
-void nx_free_buffer(nx_buffer_t *buf);
-
-
typedef enum {
NX_DEPTH_NONE,
NX_DEPTH_HEADER,
NX_DEPTH_DELIVERY_ANNOTATIONS,
NX_DEPTH_MESSAGE_ANNOTATIONS,
- NX_DEPTH_MESSAGE_PROPERTIES, // Needed for 'user-id' and 'to'
+ NX_DEPTH_PROPERTIES,
NX_DEPTH_APPLICATION_PROPERTIES,
NX_DEPTH_BODY,
NX_DEPTH_ALL
} nx_message_depth_t;
+
+typedef enum {
+ //
+ // Message Sections
+ //
+ NX_FIELD_HEADER,
+ NX_FIELD_DELIVERY_ANNOTATION,
+ NX_FIELD_MESSAGE_ANNOTATION,
+ NX_FIELD_PROPERTIES,
+ NX_FIELD_APPLICATION_PROPERTIES,
+ NX_FIELD_BODY,
+ NX_FIELD_FOOTER,
+
+ //
+ // Fields of the Header Section
+ //
+ NX_FIELD_DURABLE,
+ NX_FIELD_PRIORITY,
+ NX_FIELD_TTL,
+ NX_FIELD_FIRST_ACQUIRER,
+ NX_FIELD_DELIVERY_COUNT,
+
+ //
+ // Fields of the Properties Section
+ //
+ NX_FIELD_MESSAGE_ID,
+ NX_FIELD_USER_ID,
+ NX_FIELD_TO,
+ NX_FIELD_SUBJECT,
+ NX_FIELD_REPLY_TO,
+ NX_FIELD_CORRELATION_ID,
+ NX_FIELD_CONTENT_TYPE,
+ NX_FIELD_CONTENT_ENCODING,
+ NX_FIELD_ABSOLUTE_EXPIRY_TIME,
+ NX_FIELD_CREATION_TIME,
+ NX_FIELD_GROUP_ID,
+ NX_FIELD_GROUP_SEQUENCE,
+ NX_FIELD_REPLY_TO_GROUP_ID
+} nx_message_field_t;
+
+//
+// Functions for allocation
+//
+nx_message_t *nx_allocate_message(void);
+void nx_free_message(nx_message_t *qm);
+nx_message_t *nx_message_copy(nx_message_t *qm);
+int nx_message_persistent(nx_message_t *qm);
+int nx_message_in_memory(nx_message_t *qm);
+
+void nx_message_set_out_delivery(nx_message_t *msg, pn_delivery_t *delivery);
+pn_delivery_t *nx_message_out_delivery(nx_message_t *msg);
+void nx_message_set_in_delivery(nx_message_t *msg, pn_delivery_t *delivery);
+pn_delivery_t *nx_message_in_delivery(nx_message_t *msg);
+
//
// Functions for received messages
//
nx_message_t *nx_message_receive(pn_delivery_t *delivery);
+void nx_message_send(nx_message_t *msg, pn_link_t *link);
+
int nx_message_check(nx_message_t *msg, nx_message_depth_t depth);
-nx_field_iterator_t *nx_message_field_to(nx_message_t *msg);
-nx_field_iterator_t *nx_message_body(nx_message_t *msg);
+nx_field_iterator_t *nx_message_field_iterator(nx_message_t *msg, nx_message_field_t field);
+nx_iovec_t *nx_message_field_iovec(nx_message_t *msg, nx_message_field_t field);
+
+pn_delivery_t *nx_message_inbound_delivery(nx_message_t *qm);
//
// Functions for composed messages
//
// Convenience Functions
-void nx_message_compose_1(nx_message_t *msg, const char *to, nx_buffer_t *buf_chain);
+void nx_message_compose_1(nx_message_t *msg, const char *to, nx_buffer_list_t *buffers);
+void nx_message_copy_header(nx_message_t *msg); // Copy received header into send-header (prior to adding annotations)
+void nx_message_copy_message_annotations(nx_message_t *msg);
// Raw Functions
void nx_message_begin_header(nx_message_t *msg);
@@ -131,7 +139,7 @@ void nx_message_end_message_properties(n
void nx_message_begin_application_properties(nx_message_t *msg);
void nx_message_end_application_properties(nx_message_t *msg);
-void nx_message_append_body_data(nx_message_t *msg, nx_buffer_t *buf_chain);
+void nx_message_append_body_data(nx_message_t *msg, nx_buffer_list_t *buffers);
void nx_message_begin_body_sequence(nx_message_t *msg);
void nx_message_end_body_sequence(nx_message_t *msg);
@@ -149,14 +157,9 @@ void nx_message_insert_string(nx_message
void nx_message_insert_uuid(nx_message_t *msg, const uint8_t *value);
void nx_message_insert_symbol(nx_message_t *msg, const char *start, size_t len);
void nx_message_insert_timestamp(nx_message_t *msg, uint64_t value);
-
-//
-// Functions for buffers
-//
-unsigned char *nx_buffer_base(nx_buffer_t *buf); // Pointer to the first octet in the buffer
-unsigned char *nx_buffer_cursor(nx_buffer_t *buf); // Pointer to the first free octet in the buffer
-size_t nx_buffer_capacity(nx_buffer_t *buf); // Size of free space in the buffer in octets
-size_t nx_buffer_size(nx_buffer_t *buf); // Number of octets in the buffer
-void nx_buffer_insert(nx_buffer_t *buf, size_t len); // Notify the buffer that 'len' octets were written at cursor
+void nx_message_begin_list(nx_message_t* msg);
+void nx_message_end_list(nx_message_t* msg);
+void nx_message_begin_map(nx_message_t* msg);
+void nx_message_end_map(nx_message_t* msg);
#endif
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/alloc.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/alloc.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/alloc.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/alloc.c Mon Feb 18 10:18:24 2013
@@ -19,6 +19,7 @@
#include <qpid/nexus/alloc.h>
#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/log.h>
#include <memory.h>
#include <stdio.h>
@@ -45,9 +46,16 @@ static void nx_alloc_init(nx_alloc_type_
{
sys_mutex_lock(init_lock);
+ desc->total_size = desc->type_size;
+ if (desc->additional_size)
+ desc->total_size += *desc->additional_size;
+
+ nx_log("ALLOC", LOG_TRACE, "Initialized Allocator - type=%s type-size=%d total-size=%d",
+ desc->type_name, desc->type_size, desc->total_size);
+
if (!desc->global_pool) {
if (desc->config == 0)
- desc->config = desc->type_size > 256 ?
+ desc->config = desc->total_size > 256 ?
&nx_alloc_default_config_big : &nx_alloc_default_config_small;
assert (desc->config->local_free_list_max >= desc->config->transfer_batch_size);
@@ -121,7 +129,7 @@ void *nx_alloc(nx_alloc_type_desc_t *des
// Allocate a full batch from the heap and put it on the thread list.
//
for (idx = 0; idx < desc->config->transfer_batch_size; idx++) {
- item = (item_t*) malloc(sizeof(item_t) + desc->type_size);
+ item = (item_t*) malloc(sizeof(item_t) + desc->total_size);
if (item == 0)
break;
DEQ_ITEM_INIT(item);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/container.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/container.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/container.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/container.c Mon Feb 18 10:18:24 2013
@@ -399,10 +399,6 @@ void nx_container_initialize(void)
{
nx_log(module, LOG_TRACE, "Container Initializing");
- // TODO - move allocator init to server?
- const nx_allocator_config_t *alloc_config = nx_allocator_default_config();
- nx_allocator_initialize(alloc_config);
-
node_type_map = hash(6, 4, 1); // 64 buckets, item batches of 4
node_map = hash(10, 32, 0); // 1K buckets, item batches of 32
lock = sys_mutex();
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/iterator.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/iterator.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/iterator.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/iterator.c Mon Feb 18 10:18:24 2013
@@ -18,9 +18,9 @@
*/
#include <qpid/nexus/iterator.h>
-#include <qpid/nexus/message.h>
#include <qpid/nexus/ctools.h>
#include <qpid/nexus/alloc.h>
+#include "message_private.h"
#include <stdio.h>
#include <string.h>
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/message.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/message.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/message.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/message.c Mon Feb 18 10:18:24 2013
@@ -17,51 +17,14 @@
* under the License.
*/
-#include <qpid/nexus/message.h>
#include <qpid/nexus/ctools.h>
#include <qpid/nexus/threading.h>
+#include "message_private.h"
#include <string.h>
#include <stdio.h>
-
-//
-// Per-Thread allocator
-//
-typedef struct nx_allocator_t {
- nx_message_list_t message_free_list;
- nx_buffer_list_t buffer_free_list;
-} nx_allocator_t;
-
-//
-// Global allocator (protected by a global lock)
-//
-typedef struct {
- nx_message_list_t message_free_list;
- nx_buffer_list_t buffer_free_list;
- sys_mutex_t *lock;
-} nx_global_allocator_t;
-
-static nx_global_allocator_t global;
-static nx_allocator_config_t default_config;
-static const nx_allocator_config_t *config;
-
-
-static nx_allocator_t *nx_get_allocator(void)
-{
- static __thread nx_allocator_t *alloc = 0;
-
- if (!alloc) {
- alloc = NEW(nx_allocator_t);
-
- if (!alloc)
- return 0;
-
- DEQ_INIT(alloc->message_free_list);
- DEQ_INIT(alloc->buffer_free_list);
- }
-
- return alloc;
-}
+ALLOC_DEFINE_CONFIG(nx_message_t, sizeof(nx_message_pvt_t), 0, 0);
+ALLOC_DEFINE(nx_message_content_t);
static void advance(unsigned char **cursor, nx_buffer_t **buffer, int consume)
@@ -288,7 +251,7 @@ static int nx_check_and_advance(nx_buffe
}
-static void nx_insert(nx_message_t *msg, const uint8_t *seq, size_t len)
+static void nx_insert(nx_message_content_t *msg, const uint8_t *seq, size_t len)
{
nx_buffer_t *buf = DEQ_TAIL(msg->buffers);
@@ -312,13 +275,13 @@ static void nx_insert(nx_message_t *msg,
}
-static void nx_insert_8(nx_message_t *msg, uint8_t value)
+static void nx_insert_8(nx_message_content_t *msg, uint8_t value)
{
nx_insert(msg, &value, 1);
}
-static void nx_insert_32(nx_message_t *msg, uint32_t value)
+static void nx_insert_32(nx_message_content_t *msg, uint32_t value)
{
uint8_t buf[4];
buf[0] = (uint8_t) ((value & 0xFF000000) >> 24);
@@ -329,7 +292,7 @@ static void nx_insert_32(nx_message_t *m
}
-static void nx_insert_64(nx_message_t *msg, uint64_t value)
+static void nx_insert_64(nx_message_content_t *msg, uint64_t value)
{
uint8_t buf[8];
buf[0] = (uint8_t) ((value & 0xFF00000000000000L) >> 56);
@@ -371,7 +334,7 @@ static void nx_overwrite_32(nx_field_loc
}
-static void nx_start_list_performative(nx_message_t *msg, uint8_t code)
+static void nx_start_list_performative(nx_message_content_t *msg, uint8_t code)
{
//
// Insert the short-form performative tag
@@ -409,219 +372,172 @@ static void nx_start_list_performative(n
}
-static void nx_end_list(nx_message_t *msg)
+static void nx_end_list(nx_message_content_t *msg)
{
nx_overwrite_32(&msg->compose_length, msg->length);
nx_overwrite_32(&msg->compose_count, msg->count);
}
-const nx_allocator_config_t *nx_allocator_default_config(void)
+static nx_field_location_t *nx_message_field_location(nx_message_t *msg, nx_message_field_t field)
{
- default_config.buffer_size = 1024;
- default_config.buffer_preallocation_count = 512;
- default_config.buffer_rebalancing_batch_count = 16;
- default_config.buffer_local_storage_max = 64;
- default_config.buffer_free_list_max = 1000000;
- default_config.message_allocation_batch_count = 256;
- default_config.message_rebalancing_batch_count = 64;
- default_config.message_local_storage_max = 256;
-
- return &default_config;
-}
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ switch (field) {
+ case NX_FIELD_TO:
+ while (1) {
+ if (content->field_to.parsed)
+ return &content->field_to;
-void nx_allocator_initialize(const nx_allocator_config_t *c)
-{
- config = c;
+ if (content->section_message_properties.parsed == 0)
+ break;
- // Initialize the fields in the global structure.
- DEQ_INIT(global.message_free_list);
- DEQ_INIT(global.buffer_free_list);
- global.lock = sys_mutex();
+ nx_buffer_t *buffer = content->section_message_properties.buffer;
+ unsigned char *cursor = nx_buffer_base(buffer) + content->section_message_properties.offset;
- // Pre-allocate buffers according to the configuration
- int i;
- nx_buffer_t *buf;
+ int count = start_list(&cursor, &buffer);
+ int result;
- for (i = 0; i < config->buffer_preallocation_count; i++) {
- buf = (nx_buffer_t*) malloc (sizeof(nx_buffer_t) + config->buffer_size);
- DEQ_ITEM_INIT(buf);
- DEQ_INSERT_TAIL(global.buffer_free_list, buf);
- }
-}
+ if (count < 3)
+ break;
+ result = traverse_field(&cursor, &buffer, 0); // message_id
+ if (!result) return 0;
+ result = traverse_field(&cursor, &buffer, 0); // user_id
+ if (!result) return 0;
+ result = traverse_field(&cursor, &buffer, &content->field_to); // to
+ if (!result) return 0;
+ }
+ break;
-void nx_allocator_finalize(void)
-{
- // TODO - Free buffers and messages
-}
+ case NX_FIELD_BODY:
+ while (1) {
+ if (content->body.parsed)
+ return &content->body;
+ if (content->section_body.parsed == 0)
+ break;
-nx_message_t *nx_allocate_message(void)
-{
- nx_allocator_t *alloc = nx_get_allocator();
- nx_message_t *msg;
- int i;
+ nx_buffer_t *buffer = content->section_body.buffer;
+ unsigned char *cursor = nx_buffer_base(buffer) + content->section_body.offset;
+ int result;
- if (DEQ_SIZE(alloc->message_free_list) == 0) {
- //
- // The local free list is empty, rebalance a batch of objects from the global
- // free list.
- //
- sys_mutex_lock(global.lock);
- if (DEQ_SIZE(global.message_free_list) >= config->message_rebalancing_batch_count) {
- for (i = 0; i < config->message_rebalancing_batch_count; i++) {
- msg = DEQ_HEAD(global.message_free_list);
- DEQ_REMOVE_HEAD(global.message_free_list);
- DEQ_INSERT_TAIL(alloc->message_free_list, msg);
- }
+ result = traverse_field(&cursor, &buffer, &content->body);
+ if (!result) return 0;
}
- sys_mutex_unlock(global.lock);
- }
+ break;
- if (DEQ_SIZE(alloc->message_free_list) == 0) {
- //
- // The local free list is still empty. This means there were not enough objects on the
- // global free list to make up a batch. Allocate new objects from the heap and store
- // them in the local free list.
- //
- nx_message_t *batch = NEW_ARRAY(nx_message_t, config->message_allocation_batch_count);
- memset(batch, 0, sizeof(nx_message_t) * config->message_allocation_batch_count);
- for (i = 0; i < config->message_allocation_batch_count; i++) {
- DEQ_INSERT_TAIL(alloc->message_free_list, &batch[i]);
- }
+ default:
+ break;
}
- //
- // If the local free list is still empty, we're out of memory.
- //
- if (DEQ_SIZE(alloc->message_free_list) == 0)
+ return 0;
+}
+
+
+nx_message_t *nx_allocate_message()
+{
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) new_nx_message_t();
+ if (!msg)
return 0;
- msg = DEQ_HEAD(alloc->message_free_list);
- DEQ_REMOVE_HEAD(alloc->message_free_list);
+ DEQ_ITEM_INIT(msg);
+ msg->content = new_nx_message_content_t();
+ msg->out_delivery = 0;
- DEQ_INIT(msg->buffers);
- msg->in_delivery = NULL;
- msg->out_delivery = NULL;
- msg->section_message_header.buffer = 0;
- msg->section_message_header.parsed = 0;
- msg->section_delivery_annotation.buffer = 0;
- msg->section_delivery_annotation.parsed = 0;
- msg->section_message_annotation.buffer = 0;
- msg->section_message_annotation.parsed = 0;
- msg->section_message_properties.buffer = 0;
- msg->section_message_properties.parsed = 0;
- msg->section_application_properties.buffer = 0;
- msg->section_application_properties.parsed = 0;
- msg->section_body.buffer = 0;
- msg->section_body.parsed = 0;
- msg->section_footer.buffer = 0;
- msg->section_footer.parsed = 0;
- msg->field_user_id.buffer = 0;
- msg->field_user_id.parsed = 0;
- msg->field_to.buffer = 0;
- msg->field_to.parsed = 0;
- msg->body.buffer = 0;
- msg->body.parsed = 0;
- return msg;
+ if (msg->content == 0) {
+ free_nx_message_t((nx_message_t*) msg);
+ return 0;
+ }
+
+ memset(msg->content, 0, sizeof(nx_message_content_t));
+ msg->content->lock = sys_mutex();
+ msg->content->ref_count = 1;
+
+ return (nx_message_t*) msg;
}
-nx_buffer_t *nx_allocate_buffer(void)
-{
- nx_allocator_t *alloc = nx_get_allocator();
- nx_buffer_t *buf;
- int i;
-
- if (DEQ_SIZE(alloc->buffer_free_list) == 0) {
- sys_mutex_lock(global.lock);
- if (DEQ_SIZE(global.buffer_free_list) >= config->buffer_rebalancing_batch_count) {
- // Rebalance a batch of free descriptors to the local free list.
- for (i = 0; i < config->buffer_rebalancing_batch_count; i++) {
- buf = DEQ_HEAD(global.buffer_free_list);
- DEQ_REMOVE_HEAD(global.buffer_free_list);
- DEQ_INSERT_TAIL(alloc->buffer_free_list, buf);
- }
+void nx_free_message(nx_message_t *in_msg)
+{
+ uint32_t rc;
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) in_msg;
+ nx_message_content_t *content = msg->content;
+
+ sys_mutex_lock(content->lock);
+ rc = --content->ref_count;
+ sys_mutex_unlock(content->lock);
+
+ if (rc == 0) {
+ nx_buffer_t *buf = DEQ_HEAD(content->buffers);
+
+ while (buf) {
+ DEQ_REMOVE_HEAD(content->buffers);
+ nx_free_buffer(buf);
+ buf = DEQ_HEAD(content->buffers);
}
- sys_mutex_unlock(global.lock);
- }
- if (DEQ_SIZE(alloc->buffer_free_list) == 0) {
- // Allocate a buffer from the heap
- buf = (nx_buffer_t*) malloc (sizeof(nx_buffer_t) + config->buffer_size);
- DEQ_ITEM_INIT(buf);
- DEQ_INSERT_TAIL(alloc->buffer_free_list, buf);
+ sys_mutex_free(content->lock);
+ free_nx_message_content_t(content);
}
- if (DEQ_SIZE(alloc->buffer_free_list) == 0)
+ free_nx_message_t((nx_message_t*) msg);
+}
+
+
+nx_message_t *nx_message_copy(nx_message_t *in_msg)
+{
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) in_msg;
+ nx_message_content_t *content = msg->content;
+ nx_message_pvt_t *copy = (nx_message_pvt_t*) new_nx_message_t();
+
+ if (!copy)
return 0;
- buf = DEQ_HEAD(alloc->buffer_free_list);
- DEQ_REMOVE_HEAD(alloc->buffer_free_list);
+ DEQ_ITEM_INIT(copy);
+ copy->content = content;
+ copy->out_delivery = 0;
- buf->size = 0;
+ sys_mutex_lock(content->lock);
+ content->ref_count++;
+ sys_mutex_unlock(content->lock);
- return buf;
+ return (nx_message_t*) copy;
}
-void nx_free_message(nx_message_t *msg)
+void nx_message_set_out_delivery(nx_message_t *msg, pn_delivery_t *delivery)
{
- nx_allocator_t *alloc = nx_get_allocator();
+ ((nx_message_pvt_t*) msg)->out_delivery = delivery;
+}
- // Free any buffers in the message
- int i;
- nx_buffer_t *buf = DEQ_HEAD(msg->buffers);
- while (buf) {
- DEQ_REMOVE_HEAD(msg->buffers);
- nx_free_buffer(buf);
- buf = DEQ_HEAD(msg->buffers);
- }
- DEQ_INSERT_TAIL(alloc->message_free_list, msg);
- if (DEQ_SIZE(alloc->message_free_list) > config->message_local_storage_max) {
- //
- // The local free list has exceeded the threshold for local storage.
- // Rebalance a batch of free objects to the global free list.
- //
- sys_mutex_lock(global.lock);
- for (i = 0; i < config->message_rebalancing_batch_count; i++) {
- msg = DEQ_HEAD(alloc->message_free_list);
- DEQ_REMOVE_HEAD(alloc->message_free_list);
- DEQ_INSERT_TAIL(global.message_free_list, msg);
- }
- sys_mutex_unlock(global.lock);
- }
+pn_delivery_t *nx_message_out_delivery(nx_message_t *msg)
+{
+ return ((nx_message_pvt_t*) msg)->out_delivery;
}
-void nx_free_buffer(nx_buffer_t *buf)
+void nx_message_set_in_delivery(nx_message_t *msg, pn_delivery_t *delivery)
{
- nx_allocator_t *alloc = nx_get_allocator();
- int i;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ content->in_delivery = delivery;
+}
- DEQ_INSERT_TAIL(alloc->buffer_free_list, buf);
- if (DEQ_SIZE(alloc->buffer_free_list) > config->buffer_local_storage_max) {
- // Rebalance a batch of free descriptors to the global free list.
- sys_mutex_lock(global.lock);
- for (i = 0; i < config->buffer_rebalancing_batch_count; i++) {
- buf = DEQ_HEAD(alloc->buffer_free_list);
- DEQ_REMOVE_HEAD(alloc->buffer_free_list);
- DEQ_INSERT_TAIL(global.buffer_free_list, buf);
- }
- sys_mutex_unlock(global.lock);
- }
+
+pn_delivery_t *nx_message_in_delivery(nx_message_t *msg)
+{
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ return content->in_delivery;
}
nx_message_t *nx_message_receive(pn_delivery_t *delivery)
{
- pn_link_t *link = pn_delivery_link(delivery);
- nx_message_t *msg = (nx_message_t*) pn_delivery_get_context(delivery);
- ssize_t rc;
- nx_buffer_t *buf;
+ pn_link_t *link = pn_delivery_link(delivery);
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) pn_delivery_get_context(delivery);
+ ssize_t rc;
+ nx_buffer_t *buf;
//
// If there is no message associated with the delivery, this is the first time
@@ -629,15 +545,16 @@ nx_message_t *nx_message_receive(pn_deli
// link it and the delivery together.
//
if (!msg) {
- msg = nx_allocate_message();
+ msg = (nx_message_pvt_t*) nx_allocate_message();
pn_delivery_set_context(delivery, (void*) msg);
//
// Record the incoming delivery only if it is not settled. If it is
- // settled, there's no need to propagate disposition back to the sender.
+ // settled, it should not be recorded as no future operations on it are
+ // permitted.
//
if (!pn_delivery_settled(delivery))
- msg->in_delivery = delivery;
+ msg->content->in_delivery = delivery;
}
//
@@ -645,10 +562,10 @@ nx_message_t *nx_message_receive(pn_deli
// we will store incoming message data. If there is no buffer in the message, allocate
// an empty one and add it to the message.
//
- buf = DEQ_TAIL(msg->buffers);
+ buf = DEQ_TAIL(msg->content->buffers);
if (!buf) {
buf = nx_allocate_buffer();
- DEQ_INSERT_TAIL(msg->buffers, buf);
+ DEQ_INSERT_TAIL(msg->content->buffers, buf);
}
while (1) {
@@ -667,10 +584,10 @@ nx_message_t *nx_message_receive(pn_deli
// of the buffer size.
//
if (nx_buffer_size(buf) == 0) {
- DEQ_REMOVE_TAIL(msg->buffers);
+ DEQ_REMOVE_TAIL(msg->content->buffers);
nx_free_buffer(buf);
}
- return msg;
+ return (nx_message_t*) msg;
}
if (rc > 0) {
@@ -686,7 +603,7 @@ nx_message_t *nx_message_receive(pn_deli
//
if (nx_buffer_capacity(buf) == 0) {
buf = nx_allocate_buffer();
- DEQ_INSERT_TAIL(msg->buffers, buf);
+ DEQ_INSERT_TAIL(msg->content->buffers, buf);
}
} else
//
@@ -697,11 +614,24 @@ nx_message_t *nx_message_receive(pn_deli
break;
}
- return NULL;
+ return 0;
}
-int nx_message_check(nx_message_t *msg, nx_message_depth_t depth)
+void nx_message_send(nx_message_t *in_msg, pn_link_t *link)
+{
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) in_msg;
+ nx_buffer_t *buf = DEQ_HEAD(msg->content->buffers);
+
+ // TODO - Handle cases where annotations have been added or modified
+ while (buf) {
+ pn_link_send(link, (char*) nx_buffer_base(buf), nx_buffer_size(buf));
+ buf = DEQ_NEXT(buf);
+ }
+}
+
+
+int nx_message_check(nx_message_t *in_msg, nx_message_depth_t depth)
{
#define LONG 10
@@ -712,8 +642,8 @@ int nx_message_check(nx_message_t *msg,
#define DELIVERY_ANNOTATION_SHORT (unsigned char*) "\x00\x53\x71"
#define MESSAGE_ANNOTATION_LONG (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x72"
#define MESSAGE_ANNOTATION_SHORT (unsigned char*) "\x00\x53\x72"
-#define MESSAGE_PROPERTIES_LONG (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x73"
-#define MESSAGE_PROPERTIES_SHORT (unsigned char*) "\x00\x53\x73"
+#define PROPERTIES_LONG (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x73"
+#define PROPERTIES_SHORT (unsigned char*) "\x00\x53\x73"
#define APPLICATION_PROPERTIES_LONG (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x74"
#define APPLICATION_PROPERTIES_SHORT (unsigned char*) "\x00\x53\x74"
#define BODY_DATA_LONG (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x75"
@@ -726,8 +656,10 @@ int nx_message_check(nx_message_t *msg,
#define TAGS_MAP (unsigned char*) "\xc1\xd1"
#define TAGS_BINARY (unsigned char*) "\xa0\xb0"
- nx_buffer_t *buffer = DEQ_HEAD(msg->buffers);
- unsigned char *cursor;
+ nx_message_pvt_t *msg = (nx_message_pvt_t*) in_msg;
+ nx_message_content_t *content = msg->content;
+ nx_buffer_t *buffer = DEQ_HEAD(content->buffers);
+ unsigned char *cursor;
if (!buffer)
return 0; // Invalid - No data in the message
@@ -740,9 +672,9 @@ int nx_message_check(nx_message_t *msg,
//
// MESSAGE HEADER
//
- if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_LONG, LONG, TAGS_LIST, &msg->section_message_header))
+ if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_LONG, LONG, TAGS_LIST, &content->section_message_header))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_SHORT, SHORT, TAGS_LIST, &msg->section_message_header))
+ if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_SHORT, SHORT, TAGS_LIST, &content->section_message_header))
return 0;
if (depth == NX_DEPTH_HEADER)
@@ -751,9 +683,9 @@ int nx_message_check(nx_message_t *msg,
//
// DELIVERY ANNOTATION
//
- if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_LONG, LONG, TAGS_MAP, &msg->section_delivery_annotation))
+ if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_LONG, LONG, TAGS_MAP, &content->section_delivery_annotation))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_SHORT, SHORT, TAGS_MAP, &msg->section_delivery_annotation))
+ if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_SHORT, SHORT, TAGS_MAP, &content->section_delivery_annotation))
return 0;
if (depth == NX_DEPTH_DELIVERY_ANNOTATIONS)
@@ -762,31 +694,31 @@ int nx_message_check(nx_message_t *msg,
//
// MESSAGE ANNOTATION
//
- if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_LONG, LONG, TAGS_MAP, &msg->section_message_annotation))
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_LONG, LONG, TAGS_MAP, &content->section_message_annotation))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_SHORT, SHORT, TAGS_MAP, &msg->section_message_annotation))
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_SHORT, SHORT, TAGS_MAP, &content->section_message_annotation))
return 0;
if (depth == NX_DEPTH_MESSAGE_ANNOTATIONS)
return 1;
//
- // MESSAGE PROPERTIES
+ // PROPERTIES
//
- if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_PROPERTIES_LONG, LONG, TAGS_LIST, &msg->section_message_properties))
+ if (0 == nx_check_and_advance(&buffer, &cursor, PROPERTIES_LONG, LONG, TAGS_LIST, &content->section_message_properties))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_PROPERTIES_SHORT, SHORT, TAGS_LIST, &msg->section_message_properties))
+ if (0 == nx_check_and_advance(&buffer, &cursor, PROPERTIES_SHORT, SHORT, TAGS_LIST, &content->section_message_properties))
return 0;
- if (depth == NX_DEPTH_MESSAGE_PROPERTIES)
+ if (depth == NX_DEPTH_PROPERTIES)
return 1;
//
// APPLICATION PROPERTIES
//
- if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_LONG, LONG, TAGS_MAP, &msg->section_application_properties))
+ if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_LONG, LONG, TAGS_MAP, &content->section_application_properties))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_SHORT, SHORT, TAGS_MAP, &msg->section_application_properties))
+ if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_SHORT, SHORT, TAGS_MAP, &content->section_application_properties))
return 0;
if (depth == NX_DEPTH_APPLICATION_PROPERTIES)
@@ -795,13 +727,13 @@ int nx_message_check(nx_message_t *msg,
//
// BODY (Note that this function expects a single data section or a single AMQP sequence)
//
- if (0 == nx_check_and_advance(&buffer, &cursor, BODY_DATA_LONG, LONG, TAGS_BINARY, &msg->section_body))
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_DATA_LONG, LONG, TAGS_BINARY, &content->section_body))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, BODY_DATA_SHORT, SHORT, TAGS_BINARY, &msg->section_body))
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_DATA_SHORT, SHORT, TAGS_BINARY, &content->section_body))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_LONG, LONG, TAGS_LIST, &msg->section_body))
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_LONG, LONG, TAGS_LIST, &content->section_body))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_SHORT, SHORT, TAGS_LIST, &msg->section_body))
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_SHORT, SHORT, TAGS_LIST, &content->section_body))
return 0;
if (depth == NX_DEPTH_BODY)
@@ -810,67 +742,82 @@ int nx_message_check(nx_message_t *msg,
//
// FOOTER
//
- if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_LONG, LONG, TAGS_MAP, &msg->section_footer))
+ if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_LONG, LONG, TAGS_MAP, &content->section_footer))
return 0;
- if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_SHORT, SHORT, TAGS_MAP, &msg->section_footer))
+ if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_SHORT, SHORT, TAGS_MAP, &content->section_footer))
return 0;
return 1;
}
-nx_field_iterator_t *nx_message_field_to(nx_message_t *msg)
+nx_field_iterator_t *nx_message_field_iterator(nx_message_t *msg, nx_message_field_t field)
{
- while (1) {
- if (msg->field_to.parsed)
- return nx_field_iterator_buffer(msg->field_to.buffer, msg->field_to.offset, msg->field_to.length, ITER_VIEW_ALL);
-
- if (msg->section_message_properties.parsed == 0)
- break;
-
- nx_buffer_t *buffer = msg->section_message_properties.buffer;
- unsigned char *cursor = nx_buffer_base(buffer) + msg->section_message_properties.offset;
-
- int count = start_list(&cursor, &buffer);
- int result;
-
- if (count < 3)
- break;
-
- result = traverse_field(&cursor, &buffer, 0); // message_id
- if (!result) return 0;
- result = traverse_field(&cursor, &buffer, 0); // user_id
- if (!result) return 0;
- result = traverse_field(&cursor, &buffer, &msg->field_to); // to
- if (!result) return 0;
- }
+ nx_field_location_t *loc = nx_message_field_location(msg, field);
+ if (!loc)
+ return 0;
- return 0;
+ return nx_field_iterator_buffer(loc->buffer, loc->offset, loc->length, ITER_VIEW_ALL);
}
-nx_field_iterator_t *nx_message_body(nx_message_t *msg)
+nx_iovec_t *nx_message_field_iovec(nx_message_t *msg, nx_message_field_t field)
{
- while (1) {
- if (msg->body.parsed)
- return nx_field_iterator_buffer(msg->body.buffer, msg->body.offset, msg->body.length, ITER_VIEW_ALL);
+ nx_field_location_t *loc = nx_message_field_location(msg, field);
+ if (!loc)
+ return 0;
- if (msg->section_body.parsed == 0)
- break;
+ //
+ // Count the number of buffers this field straddles
+ //
+ int bufcnt = 1;
+ nx_buffer_t *buf = loc->buffer;
+ size_t bufsize = nx_buffer_size(buf) - loc->offset;
+ ssize_t remaining = loc->length - bufsize;
+
+ while (remaining > 0) {
+ bufcnt++;
+ buf = buf->next;
+ if (!buf)
+ return 0;
+ remaining -= nx_buffer_size(buf);
+ }
- nx_buffer_t *buffer = msg->section_body.buffer;
- unsigned char *cursor = nx_buffer_base(buffer) + msg->section_body.offset;
- int result;
+ //
+ // Allocate an iovec object big enough to hold the number of buffers
+ //
+ nx_iovec_t *iov = nx_iovec(bufcnt);
+ if (!iov)
+ return 0;
- result = traverse_field(&cursor, &buffer, &msg->body);
- if (!result) return 0;
+ //
+ // Build out the io vectors with pointers to the segments of the field in buffers
+ //
+ bufcnt = 0;
+ buf = loc->buffer;
+ bufsize = nx_buffer_size(buf) - loc->offset;
+ void *base = nx_buffer_base(buf) + loc->offset;
+ remaining = loc->length;
+
+ while (remaining > 0) {
+ nx_iovec_array(iov)[bufcnt].iov_base = base;
+ nx_iovec_array(iov)[bufcnt].iov_len = bufsize;
+ bufcnt++;
+ remaining -= bufsize;
+ if (remaining > 0) {
+ buf = buf->next;
+ base = nx_buffer_base(buf);
+ bufsize = nx_buffer_size(buf);
+ if (bufsize > remaining)
+ bufsize = remaining;
+ }
}
- return 0;
+ return iov;
}
-void nx_message_compose_1(nx_message_t *msg, const char *to, nx_buffer_t *buf_chain)
+void nx_message_compose_1(nx_message_t *msg, const char *to, nx_buffer_list_t *buffers)
{
nx_message_begin_header(msg);
nx_message_insert_boolean(msg, 0); // durable
@@ -896,20 +843,20 @@ void nx_message_compose_1(nx_message_t *
//nx_message_insert_null(msg); // reply-to-group-id
nx_message_end_message_properties(msg);
- if (buf_chain)
- nx_message_append_body_data(msg, buf_chain);
+ if (buffers)
+ nx_message_append_body_data(msg, buffers);
}
void nx_message_begin_header(nx_message_t *msg)
{
- nx_start_list_performative(msg, 0x70);
+ nx_start_list_performative(MSG_CONTENT(msg), 0x70);
}
void nx_message_end_header(nx_message_t *msg)
{
- nx_end_list(msg);
+ nx_end_list(MSG_CONTENT(msg));
}
@@ -939,13 +886,13 @@ void nx_message_end_message_annotations(
void nx_message_begin_message_properties(nx_message_t *msg)
{
- nx_start_list_performative(msg, 0x73);
+ nx_start_list_performative(MSG_CONTENT(msg), 0x73);
}
void nx_message_end_message_properties(nx_message_t *msg)
{
- nx_end_list(msg);
+ nx_end_list(MSG_CONTENT(msg));
}
@@ -961,34 +908,40 @@ void nx_message_end_application_properti
}
-void nx_message_append_body_data(nx_message_t *msg, nx_buffer_t *buf_chain)
+void nx_message_append_body_data(nx_message_t *msg, nx_buffer_list_t *buffers)
{
- uint32_t len = 0;
- nx_buffer_t *buf = buf_chain;
- nx_buffer_t *last = 0;
- size_t count = 0;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ nx_buffer_t *buf = DEQ_HEAD(*buffers);
+ uint32_t len = 0;
+ //
+ // Calculate the size of the body to be appended.
+ //
while (buf) {
len += nx_buffer_size(buf);
- count++;
- last = buf;
buf = DEQ_NEXT(buf);
}
- nx_insert(msg, (const uint8_t*) "\x00\x53\x75", 3);
+ //
+ // Insert a DATA section performative header.
+ //
+ nx_insert(content, (const uint8_t*) "\x00\x53\x75", 3);
if (len < 256) {
- nx_insert_8(msg, 0xa0); // vbin8
- nx_insert_8(msg, (uint8_t) len);
+ nx_insert_8(content, 0xa0); // vbin8
+ nx_insert_8(content, (uint8_t) len);
} else {
- nx_insert_8(msg, 0xb0); // vbin32
- nx_insert_32(msg, len);
+ nx_insert_8(content, 0xb0); // vbin32
+ nx_insert_32(content, len);
}
- if (len > 0) {
- buf_chain->prev = msg->buffers.tail;
- msg->buffers.tail->next = buf_chain;
- msg->buffers.tail = last;
- msg->buffers.size += count;
+ //
+ // Move the supplied buffers to the tail of the message's buffer list.
+ //
+ buf = DEQ_HEAD(*buffers);
+ while (buf) {
+ DEQ_REMOVE_HEAD(*buffers);
+ DEQ_INSERT_TAIL(content->buffers, buf);
+ buf = DEQ_HEAD(*buffers);
}
}
@@ -1017,148 +970,151 @@ void nx_message_end_footer(nx_message_t
void nx_message_insert_null(nx_message_t *msg)
{
- nx_insert_8(msg, 0x40);
- msg->count++;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ nx_insert_8(content, 0x40);
+ content->count++;
}
void nx_message_insert_boolean(nx_message_t *msg, int value)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
if (value)
- nx_insert(msg, (const uint8_t*) "\x56\x01", 2);
+ nx_insert(content, (const uint8_t*) "\x56\x01", 2);
else
- nx_insert(msg, (const uint8_t*) "\x56\x00", 2);
- msg->count++;
+ nx_insert(content, (const uint8_t*) "\x56\x00", 2);
+ content->count++;
}
void nx_message_insert_ubyte(nx_message_t *msg, uint8_t value)
{
- nx_insert_8(msg, 0x50);
- nx_insert_8(msg, value);
- msg->count++;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ nx_insert_8(content, 0x50);
+ nx_insert_8(content, value);
+ content->count++;
}
void nx_message_insert_uint(nx_message_t *msg, uint32_t value)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
if (value == 0) {
- nx_insert_8(msg, 0x43); // uint0
+ nx_insert_8(content, 0x43); // uint0
} else if (value < 256) {
- nx_insert_8(msg, 0x52); // smalluint
- nx_insert_8(msg, (uint8_t) value);
+ nx_insert_8(content, 0x52); // smalluint
+ nx_insert_8(content, (uint8_t) value);
} else {
- nx_insert_8(msg, 0x70); // uint
- nx_insert_32(msg, value);
+ nx_insert_8(content, 0x70); // uint
+ nx_insert_32(content, value);
}
- msg->count++;
+ content->count++;
}
void nx_message_insert_ulong(nx_message_t *msg, uint64_t value)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
if (value == 0) {
- nx_insert_8(msg, 0x44); // ulong0
+ nx_insert_8(content, 0x44); // ulong0
} else if (value < 256) {
- nx_insert_8(msg, 0x53); // smallulong
- nx_insert_8(msg, (uint8_t) value);
+ nx_insert_8(content, 0x53); // smallulong
+ nx_insert_8(content, (uint8_t) value);
} else {
- nx_insert_8(msg, 0x80); // ulong
- nx_insert_64(msg, value);
+ nx_insert_8(content, 0x80); // ulong
+ nx_insert_64(content, value);
}
- msg->count++;
+ content->count++;
}
void nx_message_insert_binary(nx_message_t *msg, const uint8_t *start, size_t len)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
if (len < 256) {
- nx_insert_8(msg, 0xa0); // vbin8
- nx_insert_8(msg, (uint8_t) len);
+ nx_insert_8(content, 0xa0); // vbin8
+ nx_insert_8(content, (uint8_t) len);
} else {
- nx_insert_8(msg, 0xb0); // vbin32
- nx_insert_32(msg, len);
+ nx_insert_8(content, 0xb0); // vbin32
+ nx_insert_32(content, len);
}
- nx_insert(msg, start, len);
- msg->count++;
+ nx_insert(content, start, len);
+ content->count++;
}
void nx_message_insert_string(nx_message_t *msg, const char *start)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
uint32_t len = strlen(start);
if (len < 256) {
- nx_insert_8(msg, 0xa1); // str8-utf8
- nx_insert_8(msg, (uint8_t) len);
- nx_insert(msg, (const uint8_t*) start, len);
+ nx_insert_8(content, 0xa1); // str8-utf8
+ nx_insert_8(content, (uint8_t) len);
+ nx_insert(content, (const uint8_t*) start, len);
} else {
- nx_insert_8(msg, 0xb1); // str32-utf8
- nx_insert_32(msg, len);
- nx_insert(msg, (const uint8_t*) start, len);
+ nx_insert_8(content, 0xb1); // str32-utf8
+ nx_insert_32(content, len);
+ nx_insert(content, (const uint8_t*) start, len);
}
- msg->count++;
+ content->count++;
}
void nx_message_insert_uuid(nx_message_t *msg, const uint8_t *value)
{
- nx_insert_8(msg, 0x98); // uuid
- nx_insert(msg, value, 16);
- msg->count++;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ nx_insert_8(content, 0x98); // uuid
+ nx_insert(content, value, 16);
+ content->count++;
}
void nx_message_insert_symbol(nx_message_t *msg, const char *start, size_t len)
{
+ nx_message_content_t *content = MSG_CONTENT(msg);
if (len < 256) {
- nx_insert_8(msg, 0xa3); // sym8
- nx_insert_8(msg, (uint8_t) len);
- nx_insert(msg, (const uint8_t*) start, len);
+ nx_insert_8(content, 0xa3); // sym8
+ nx_insert_8(content, (uint8_t) len);
+ nx_insert(content, (const uint8_t*) start, len);
} else {
- nx_insert_8(msg, 0xb3); // sym32
- nx_insert_32(msg, len);
- nx_insert(msg, (const uint8_t*) start, len);
+ nx_insert_8(content, 0xb3); // sym32
+ nx_insert_32(content, len);
+ nx_insert(content, (const uint8_t*) start, len);
}
- msg->count++;
+ content->count++;
}
void nx_message_insert_timestamp(nx_message_t *msg, uint64_t value)
{
- nx_insert_8(msg, 0x83); // timestamp
- nx_insert_64(msg, value);
- msg->count++;
+ nx_message_content_t *content = MSG_CONTENT(msg);
+ nx_insert_8(content, 0x83); // timestamp
+ nx_insert_64(content, value);
+ content->count++;
}
-unsigned char *nx_buffer_base(nx_buffer_t *buf)
+void nx_message_begin_list(nx_message_t* msg)
{
- return (unsigned char*) &buf[1];
-}
-
-
-unsigned char *nx_buffer_cursor(nx_buffer_t *buf)
-{
- return ((unsigned char*) &buf[1]) + buf->size;
+ assert(0); // Not Implemented
}
-size_t nx_buffer_capacity(nx_buffer_t *buf)
+void nx_message_end_list(nx_message_t* msg)
{
- return config->buffer_size - buf->size;
+ assert(0); // Not Implemented
}
-size_t nx_buffer_size(nx_buffer_t *buf)
+void nx_message_begin_map(nx_message_t* msg)
{
- return buf->size;
+ assert(0); // Not Implemented
}
-void nx_buffer_insert(nx_buffer_t *buf, size_t len)
+void nx_message_end_map(nx_message_t* msg)
{
- buf->size += len;
- assert(buf->size <= config->buffer_size);
+ assert(0); // Not Implemented
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/timer.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/timer.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/timer.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/src/timer.c Mon Feb 18 10:18:24 2013
@@ -21,15 +21,17 @@
#include "server_private.h"
#include <qpid/nexus/ctools.h>
#include <qpid/nexus/threading.h>
+#include <qpid/nexus/alloc.h>
#include <assert.h>
#include <stdio.h>
static sys_mutex_t *lock;
-static nx_timer_list_t free_list;
static nx_timer_list_t idle_timers;
static nx_timer_list_t scheduled_timers;
static long time_base;
+ALLOC_DECLARE(nx_timer_t);
+ALLOC_DEFINE(nx_timer_t);
//=========================================================================
// Private static functions
@@ -67,27 +69,21 @@ static void nx_timer_cancel_LH(nx_timer_
nx_timer_t *nx_timer(nx_timer_cb_t cb, void* context)
{
- nx_timer_t *timer;
+ nx_timer_t *timer = new_nx_timer_t();
+ if (!timer)
+ return 0;
+
+ DEQ_ITEM_INIT(timer);
+
+ timer->handler = cb;
+ timer->context = context;
+ timer->delta_time = 0;
+ timer->state = TIMER_IDLE;
sys_mutex_lock(lock);
-
- timer = DEQ_HEAD(free_list);
- if (timer) {
- DEQ_REMOVE_HEAD(free_list);
- } else {
- timer = NEW(nx_timer_t);
- DEQ_ITEM_INIT(timer);
- }
-
- if (timer) {
- timer->handler = cb;
- timer->context = context;
- timer->delta_time = 0;
- timer->state = TIMER_IDLE;
- DEQ_INSERT_TAIL(idle_timers, timer);
- }
-
+ DEQ_INSERT_TAIL(idle_timers, timer);
sys_mutex_unlock(lock);
+
return timer;
}
@@ -97,9 +93,10 @@ void nx_timer_free(nx_timer_t *timer)
sys_mutex_lock(lock);
nx_timer_cancel_LH(timer);
DEQ_REMOVE(idle_timers, timer);
- DEQ_INSERT_TAIL(free_list, timer);
- timer->state = TIMER_FREE;
sys_mutex_unlock(lock);
+
+ timer->state = TIMER_FREE;
+ free_nx_timer_t(timer);
}
@@ -180,7 +177,6 @@ void nx_timer_cancel(nx_timer_t *timer)
void nx_timer_initialize(sys_mutex_t *server_lock)
{
lock = server_lock;
- DEQ_INIT(free_list);
DEQ_INIT(idle_timers);
DEQ_INIT(scheduled_timers);
time_base = 0;
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/alloc_test.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/alloc_test.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/alloc_test.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/alloc_test.c Mon Feb 18 10:18:24 2013
@@ -30,7 +30,7 @@ typedef struct {
nx_alloc_config_t config = {3, 7, 10};
ALLOC_DECLARE(object_t);
-ALLOC_DEFINE_CONFIG(object_t, &config);
+ALLOC_DEFINE_CONFIG(object_t, sizeof(object_t), 0, &config);
static char* check_stats(nx_alloc_stats_t *stats, uint64_t ah, uint64_t fh, uint64_t ht, uint64_t rt, uint64_t rg)
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/message_test.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/message_test.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/message_test.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/message_test.c Mon Feb 18 10:18:24 2013
@@ -20,26 +20,18 @@
#include "test_case.h"
#include <stdio.h>
#include <string.h>
-#include <qpid/nexus/message.h>
+#include "message_private.h"
#include <qpid/nexus/iterator.h>
#include <proton/message.h>
-static char* test_init(void *context)
-{
- nx_allocator_initialize(nx_allocator_default_config());
- nx_allocator_finalize();
- return 0;
-}
-
-
static char* test_send_to_messenger(void *context)
{
- nx_allocator_initialize(nx_allocator_default_config());
+ nx_message_t *msg = nx_allocate_message();
+ nx_message_content_t *content = MSG_CONTENT(msg);
- nx_message_t *msg = nx_allocate_message();
nx_message_compose_1(msg, "test_addr_0", 0);
- nx_buffer_t *buf = DEQ_HEAD(msg->buffers);
+ nx_buffer_t *buf = DEQ_HEAD(content->buffers);
if (buf == 0) return "Expected a buffer in the test message";
pn_message_t *pn_msg = pn_message();
@@ -52,15 +44,12 @@ static char* test_send_to_messenger(void
pn_message_free(pn_msg);
nx_free_message(msg);
- nx_allocator_finalize();
return 0;
}
static char* test_receive_from_messenger(void *context)
{
- nx_allocator_initialize(nx_allocator_default_config());
-
pn_message_t *pn_msg = pn_message();
pn_message_set_address(pn_msg, "test_addr_1");
@@ -70,12 +59,14 @@ static char* test_receive_from_messenger
if (result != 0) return "Error in pn_message_encode";
nx_buffer_insert(buf, size);
- nx_message_t *msg = nx_allocate_message();
- DEQ_INSERT_TAIL(msg->buffers, buf);
+ nx_message_t *msg = nx_allocate_message();
+ nx_message_content_t *content = MSG_CONTENT(msg);
+
+ DEQ_INSERT_TAIL(content->buffers, buf);
int valid = nx_message_check(msg, NX_DEPTH_ALL);
if (!valid) return "nx_message_check returns 'invalid'";
- nx_field_iterator_t *iter = nx_message_field_to(msg);
+ nx_field_iterator_t *iter = nx_message_field_iterator(msg, NX_FIELD_TO);
if (iter == 0) return "Expected an iterator for the 'to' field";
if (!nx_field_iterator_equal(iter, (unsigned char*) "test_addr_1"))
@@ -84,15 +75,12 @@ static char* test_receive_from_messenger
pn_message_free(pn_msg);
nx_free_message(msg);
- nx_allocator_finalize();
return 0;
}
static char* test_insufficient_check_depth(void *context)
{
- nx_allocator_initialize(nx_allocator_default_config());
-
pn_message_t *pn_msg = pn_message();
pn_message_set_address(pn_msg, "test_addr_2");
@@ -102,17 +90,18 @@ static char* test_insufficient_check_dep
if (result != 0) return "Error in pn_message_encode";
nx_buffer_insert(buf, size);
- nx_message_t *msg = nx_allocate_message();
- DEQ_INSERT_TAIL(msg->buffers, buf);
+ nx_message_t *msg = nx_allocate_message();
+ nx_message_content_t *content = MSG_CONTENT(msg);
+
+ DEQ_INSERT_TAIL(content->buffers, buf);
int valid = nx_message_check(msg, NX_DEPTH_DELIVERY_ANNOTATIONS);
if (!valid) return "nx_message_check returns 'invalid'";
- nx_field_iterator_t *iter = nx_message_field_to(msg);
+ nx_field_iterator_t *iter = nx_message_field_iterator(msg, NX_FIELD_TO);
if (iter) return "Expected no iterator for the 'to' field";
nx_free_message(msg);
- nx_allocator_finalize();
return 0;
}
@@ -121,7 +110,6 @@ int message_tests(void)
{
int result = 0;
- TEST_CASE(test_init, 0);
TEST_CASE(test_send_to_messenger, 0);
TEST_CASE(test_receive_from_messenger, 0);
TEST_CASE(test_insufficient_check_depth, 0);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/timer_test.c
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/timer_test.c?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/timer_test.c (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/nexus/tests/timer_test.c Mon Feb 18 10:18:24 2013
@@ -19,6 +19,7 @@
#include <stdio.h>
#include <qpid/nexus/timer.h>
+#include "alloc_private.h"
#include "timer_private.h"
#include "test_case.h"
#include <qpid/nexus/threading.h>
@@ -341,6 +342,7 @@ static char* test_big(void *context)
int timer_tests(void)
{
int result = 0;
+ nx_alloc_initialize();
fire_mask = 0;
DEQ_INIT(pending_timers);
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/extras/qmf/src/py/qmf/console.py
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/extras/qmf/src/py/qmf/console.py?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/extras/qmf/src/py/qmf/console.py (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/extras/qmf/src/py/qmf/console.py Mon Feb 18 10:18:24 2013
@@ -2432,10 +2432,13 @@ class Broker(Thread):
if sys.version_info[:2] < (2, 6): # 2.6+ uses openssl - it's ok
force_blocking = True
sock.setblocking(1)
+ certfile = None
if 'ssl_certfile' in self.connectArgs:
- connSock = ssl(sock, certfile=self.connectArgs['ssl_certfile'])
- else:
- connSock = ssl(sock)
+ certfile = self.connectArgs['ssl_certfile']
+ keyfile = None
+ if 'ssl_keyfile' in self.connectArgs:
+ keyfile = self.connectArgs['ssl_keyfile']
+ connSock = ssl(sock, certfile=certfile, keyfile=keyfile)
else:
connSock = sock
self.conn = Connection(connSock, username=self.authUser, password=self.authPass,
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/.gitignore
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/.gitignore?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/.gitignore (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/.gitignore Mon Feb 18 10:18:24 2013
@@ -18,3 +18,4 @@
#
*.swp
eclipse-projects/*
+derby.log
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-client/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client:r1438054-1446845
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-client-jms/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-client-jms:r1438054-1446845
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-common/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/amqp-1-0-common:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/framing/AMQFrame.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/framing/AMQFrame.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/framing/AMQFrame.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/amqp-1-0-common/src/main/java/org/apache/qpid/amqp_1_0/framing/AMQFrame.java Mon Feb 18 10:18:24 2013
@@ -21,6 +21,7 @@
package org.apache.qpid.amqp_1_0.framing;
+import org.apache.qpid.amqp_1_0.type.Binary;
import org.apache.qpid.amqp_1_0.type.FrameBody;
import java.nio.ByteBuffer;
@@ -65,4 +66,11 @@ public abstract class AMQFrame<T>
return _frameBody;
}
+ @Override
+ public String toString()
+ {
+ return "AMQFrame{" +
+ "frameBody=" + _frameBody +
+ '}';
+ }
}
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/AbstractBDBMessageStore.java Mon Feb 18 10:18:24 2013
@@ -1534,7 +1534,8 @@ public abstract class AbstractBDBMessage
else
{
ByteBuffer buf = ByteBuffer.allocate(size);
- getContent(offsetInMessage, buf);
+ int length = getContent(offsetInMessage, buf);
+ buf.limit(length);
buf.position(0);
return buf;
}
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker:r1438054-1446845
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/bin/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/bin:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java Mon Feb 18 10:18:24 2013
@@ -342,6 +342,14 @@ public class MessageMetaData_1_0 impleme
{
private final AMQPDescribedTypeRegistry _typeRegistry = AMQPDescribedTypeRegistry.newInstance();
+ private MetaDataFactory()
+ {
+ _typeRegistry.registerTransportLayer();
+ _typeRegistry.registerMessagingLayer();
+ _typeRegistry.registerTransactionLayer();
+ _typeRegistry.registerSecurityLayer();
+ }
+
public MessageMetaData_1_0 createMetaData(ByteBuffer buf)
{
ValueHandler valueHandler = new ValueHandler(_typeRegistry);
@@ -354,7 +362,8 @@ public class MessageMetaData_1_0 impleme
try
{
ByteBuffer encodedBuf = buf.duplicate();
- sections.add((Section) valueHandler.parse(buf));
+ Object parse = valueHandler.parse(buf);
+ sections.add((Section) parse);
encodedBuf.limit(buf.position());
encodedSections.add(encodedBuf);
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/message/MessageMetaData_1_0.java:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java Mon Feb 18 10:18:24 2013
@@ -146,12 +146,13 @@ public class ProtocolEngine_1_0_0 implem
Container container = new Container(_broker.getId().toString());
+ VirtualHost virtualHost = _broker.getVirtualHostRegistry().getVirtualHost((String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST));
+
_conn = new ConnectionEndpoint(container, asSaslServerProvider(_broker.getSubjectCreator(
getLocalAddress())));
- VirtualHost virtualHost = _broker.getVirtualHostRegistry().getVirtualHost((String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST));
+ _conn.setRemoteAddress(_network.getRemoteAddress());
_conn.setConnectionEventListener(new Connection_1_0(virtualHost, _conn, _connectionId));
_conn.setFrameOutputHandler(this);
- _conn.setRemoteAddress(_network.getRemoteAddress());
_frameWriter = new FrameWriter(_conn.getDescribedTypeRegistry());
_frameHandler = new FrameHandler(_conn);
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0.java:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java Mon Feb 18 10:18:24 2013
@@ -161,12 +161,10 @@ public class ProtocolEngine_1_0_0_SASL i
Container container = new Container(_broker.getId().toString());
- _conn = new ConnectionEndpoint(container, asSaslServerProvider(_broker.getSubjectCreator(getLocalAddress())));
VirtualHost virtualHost = _broker.getVirtualHostRegistry().getVirtualHost((String)_broker.getAttribute(Broker.DEFAULT_VIRTUAL_HOST));
- _conn.setConnectionEventListener(new Connection_1_0(virtualHost, _conn, _connectionId));
+ _conn = new ConnectionEndpoint(container, asSaslServerProvider(_broker.getSubjectCreator(getLocalAddress())));
_conn.setRemoteAddress(getRemoteAddress());
-
-
+ _conn.setConnectionEventListener(new Connection_1_0(virtualHost, _conn, _connectionId));
_conn.setFrameOutputHandler(this);
_conn.setSaslFrameOutput(this);
@@ -372,11 +370,8 @@ public class ProtocolEngine_1_0_0_SASL i
FRAME_LOGGER.fine("SEND[" + getRemoteAddress() + "|" + amqFrame.getChannel() + "] : " + amqFrame.getFrameBody());
}
-
_frameWriter.setValue(amqFrame);
-
-
ByteBuffer dup = ByteBuffer.allocate(_conn.getMaxFrameSize());
int size = _frameWriter.writeToBuffer(dup);
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/ProtocolEngine_1_0_0_SASL.java:r1438054-1446845
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/protocol/v1_0/Message_1_0.java Mon Feb 18 10:18:24 2013
@@ -23,7 +23,9 @@ package org.apache.qpid.server.protocol.
import java.lang.ref.WeakReference;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.server.message.InboundMessage;
import org.apache.qpid.server.message.MessageMetaData_1_0;
@@ -33,11 +35,45 @@ import org.apache.qpid.server.store.Stor
public class Message_1_0 implements ServerMessage, InboundMessage
{
+
+
+ private static final AtomicIntegerFieldUpdater<Message_1_0> _refCountUpdater =
+ AtomicIntegerFieldUpdater.newUpdater(Message_1_0.class, "_referenceCount");
+
+ private volatile int _referenceCount = 0;
+
private final StoredMessage<MessageMetaData_1_0> _storedMessage;
private List<ByteBuffer> _fragments;
private WeakReference<Session_1_0> _session;
+ public Message_1_0(final StoredMessage<MessageMetaData_1_0> storedMessage)
+ {
+ _storedMessage = storedMessage;
+ _session = null;
+ _fragments = restoreFragments(storedMessage);
+ }
+
+ private static List<ByteBuffer> restoreFragments(StoredMessage<MessageMetaData_1_0> storedMessage)
+ {
+ ArrayList<ByteBuffer> fragments = new ArrayList<ByteBuffer>();
+ final int FRAGMENT_SIZE = 2048;
+ int offset = 0;
+ ByteBuffer b;
+ do
+ {
+
+ b = storedMessage.getContent(offset,FRAGMENT_SIZE);
+ if(b.hasRemaining())
+ {
+ fragments.add(b);
+ offset+= b.remaining();
+ }
+ }
+ while(b.hasRemaining());
+ return fragments;
+ }
+
public Message_1_0(final StoredMessage<MessageMetaData_1_0> storedMessage,
final List<ByteBuffer> fragments,
final Session_1_0 session)
@@ -142,7 +178,61 @@ public class Message_1_0 implements Serv
public Session_1_0 getSession()
{
- return _session.get();
+ return _session == null ? null : _session.get();
+ }
+
+
+ public boolean incrementReference()
+ {
+ if(_refCountUpdater.incrementAndGet(this) <= 0)
+ {
+ _refCountUpdater.decrementAndGet(this);
+ return false;
+ }
+ else
+ {
+ return true;
+ }
+ }
+
+ /**
+ * Threadsafe. This will decrement the reference count and when it reaches zero will remove the message from the
+ * message store.
+ *
+ *
+ * @throws org.apache.qpid.server.queue.MessageCleanupException when an attempt was made to remove the message from the message store and that
+ * failed
+ */
+ public void decrementReference()
+ {
+ int count = _refCountUpdater.decrementAndGet(this);
+
+ // note that the operation of decrementing the reference count and then removing the message does not
+ // have to be atomic since the ref count starts at 1 and the exchange itself decrements that after
+ // the message has been passed to all queues. i.e. we are
+ // not relying on the all the increments having taken place before the delivery manager decrements.
+ if (count == 0)
+ {
+ // set the reference count way below 0 so that we can detect that the message has been deleted
+ // this is to guard against the message being spontaneously recreated (from the mgmt console)
+ // by copying from other queues at the same time as it is being removed.
+ _refCountUpdater.set(this,Integer.MIN_VALUE/2);
+
+ // must check if the handle is null since there may be cases where we decide to throw away a message
+ // and the handle has not yet been constructed
+ if (_storedMessage != null)
+ {
+ _storedMessage.remove();
+ }
+ }
+ else
+ {
+ if (count < 0)
+ {
+ throw new RuntimeException("Reference count for message id " + getMessageNumber()
+ + " has gone below 0.");
+ }
+ }
}
public static class Reference extends MessageReference<Message_1_0>
@@ -154,13 +244,13 @@ public class Message_1_0 implements Serv
protected void onReference(Message_1_0 message)
{
-
+ message.incrementReference();
}
protected void onRelease(Message_1_0 message)
{
-
+ message.decrementReference();
}
-}
+ }
}
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/store/derby/DerbyMessageStore.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/store/derby/DerbyMessageStore.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/store/derby/DerbyMessageStore.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/store/derby/DerbyMessageStore.java Mon Feb 18 10:18:24 2013
@@ -1793,8 +1793,9 @@ public class DerbyMessageStore implement
public ByteBuffer getContent(int offsetInMessage, int size)
{
ByteBuffer buf = ByteBuffer.allocate(size);
- getContent(offsetInMessage, buf);
+ int length = getContent(offsetInMessage, buf);
buf.position(0);
+ buf.limit(length);
return buf;
}
Propchange: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost:r1438054-1446845
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHostConfigRecoveryHandler.java Mon Feb 18 10:18:24 2013
@@ -41,8 +41,10 @@ import org.apache.qpid.server.logging.su
import org.apache.qpid.server.message.AMQMessage;
import org.apache.qpid.server.message.AbstractServerMessageImpl;
import org.apache.qpid.server.message.EnqueableMessage;
+import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.server.message.MessageTransferMessage;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.protocol.v1_0.Message_1_0;
import org.apache.qpid.server.queue.AMQQueue;
import org.apache.qpid.server.queue.AMQQueueFactory;
import org.apache.qpid.server.queue.QueueEntry;
@@ -75,7 +77,7 @@ public class VirtualHostConfigRecoveryHa
private final VirtualHost _virtualHost;
private final Map<String, Integer> _queueRecoveries = new TreeMap<String, Integer>();
- private final Map<Long, AbstractServerMessageImpl> _recoveredMessages = new HashMap<Long, AbstractServerMessageImpl>();
+ private final Map<Long, ServerMessage> _recoveredMessages = new HashMap<Long, ServerMessage>();
private final Map<Long, StoredMessage> _unusedMessages = new HashMap<Long, StoredMessage>();
private MessageStoreLogSubject _logSubject;
@@ -167,7 +169,7 @@ public class VirtualHostConfigRecoveryHa
public void message(StoredMessage message)
{
- AbstractServerMessageImpl serverMessage;
+ ServerMessage serverMessage;
switch(message.getMetaData().getType())
{
case META_DATA_0_8:
@@ -176,6 +178,9 @@ public class VirtualHostConfigRecoveryHa
case META_DATA_0_10:
serverMessage = new MessageTransferMessage(message, null);
break;
+ case META_DATA_1_0:
+ serverMessage = new Message_1_0(message);
+ break;
default:
throw new RuntimeException("Unknown message type retrieved from store " + message.getMetaData().getClass());
}
@@ -206,12 +211,13 @@ public class VirtualHostConfigRecoveryHa
if(queue != null)
{
final long messageId = record.getMessage().getMessageNumber();
- final AbstractServerMessageImpl message = _recoveredMessages.get(messageId);
+ final ServerMessage message = _recoveredMessages.get(messageId);
_unusedMessages.remove(messageId);
if(message != null)
{
- message.incrementReference();
+ final MessageReference ref = message.newReference();
+
branch.enqueue(queue,message);
@@ -224,7 +230,7 @@ public class VirtualHostConfigRecoveryHa
{
queue.enqueue(message, true, null);
- message.decrementReference();
+ ref.release();
}
catch (AMQException e)
{
@@ -236,7 +242,7 @@ public class VirtualHostConfigRecoveryHa
public void onRollback()
{
- message.decrementReference();
+ ref.release();
}
});
}
@@ -265,7 +271,7 @@ public class VirtualHostConfigRecoveryHa
if(queue != null)
{
final long messageId = record.getMessage().getMessageNumber();
- final AbstractServerMessageImpl message = _recoveredMessages.get(messageId);
+ final ServerMessage message = _recoveredMessages.get(messageId);
_unusedMessages.remove(messageId);
if(message != null)
Modified: qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java?rev=1447183&r1=1447182&r2=1447183&view=diff
==============================================================================
--- qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java (original)
+++ qpid/branches/java-broker-config-qpid-4390/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java Mon Feb 18 10:18:24 2013
@@ -27,6 +27,7 @@ import org.apache.qpid.framing.ContentHe
import org.apache.qpid.framing.abstraction.MessagePublishInfo;
import org.apache.qpid.server.message.AMQMessage;
import org.apache.qpid.server.message.MessageMetaData;
+import org.apache.qpid.server.message.MessageReference;
import org.apache.qpid.test.utils.QpidTestCase;
/**
@@ -86,10 +87,12 @@ public class ReferenceCountingTest exten
AMQMessage message = new AMQMessage(storedMessage);
- message.incrementReference();
+ MessageReference ref = message.newReference();
assertEquals(1, _store.getMessageCount());
- message.decrementReference();
+
+ ref.release();
+
assertEquals(0, _store.getMessageCount());
}
@@ -142,13 +145,13 @@ public class ReferenceCountingTest exten
AMQMessage message = new AMQMessage(storedMessage);
- message.incrementReference();
+ MessageReference ref = message.newReference();
// we call routing complete to set up the handle
// message.routingComplete(_store, _storeContext, new MessageHandleFactory());
assertEquals(1, _store.getMessageCount());
- message.incrementReference();
- message.decrementReference();
+ MessageReference ref2 = message.newReference();
+ ref.release();
assertEquals(1, _store.getMessageCount());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org