NO-JIRA - Additional Development

- Added buffer of saved log messages for remote retrieval.
- Added __FILE__ and __LINE__ annotations to logs.
- Refactored dx_message_check() so it can be called multiple times with different depths.
- Separated the buffer-size-specific tests into a separate unit test executable.


git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1478538 13f79535-47bb-0310-9956-ffa450edef68

11 files changed, 263 insertions, 97 deletions

diff --git a/qpid/extras/dispatch/src/agent.c b/qpid/extras/dispatch/src/agent.c
index aca9ab3560..08425cee24 100644
--- a/qpid/extras/dispatch/src/agent.c
+++ b/qpid/extras/dispatch/src/agent.c
@@ -59,18 +59,44 @@ ALLOC_DECLARE(dx_agent_request_t);
 ALLOC_DEFINE(dx_agent_request_t);
 
 
+static void dx_agent_process_request(dx_message_t *msg)
+{
+    if (!dx_message_check(msg, DX_DEPTH_BODY))
+        return;
+    printf("Processing Agent Request\n");
+}
+
+
 static void dx_agent_timer_handler(void *context)
 {
-    // TODO - Process the in_fifo here
+    dx_agent_t   *agent = (dx_agent_t*) context;
+    dx_message_t *msg;
+
+    do {
+        sys_mutex_lock(agent->lock);
+        msg = DEQ_HEAD(agent->in_fifo);
+        if (msg)
+            DEQ_REMOVE_HEAD(agent->in_fifo);
+        sys_mutex_unlock(agent->lock);
+
+        if (msg) {
+            dx_agent_process_request(msg);
+            dx_free_message(msg);
+        }
+    } while (msg);
 }
 
 
 static void dx_agent_rx_handler(void *context, dx_message_t *msg)
 {
-    dx_agent_t *agent = (dx_agent_t*) context;
-    DEQ_INSERT_TAIL(agent->in_fifo, msg);
+    dx_agent_t   *agent = (dx_agent_t*) context;
+    dx_message_t *copy  = dx_message_copy(msg);
+
+    sys_mutex_lock(agent->lock);
+    DEQ_INSERT_TAIL(agent->in_fifo, copy);
+    sys_mutex_unlock(agent->lock);
+
     dx_timer_schedule(agent->timer, 0);
-    printf("dx_agent_rx_handler - inbound message\n");
 }
 
 
diff --git a/qpid/extras/dispatch/src/dispatch.c b/qpid/extras/dispatch/src/dispatch.c
index 47a1a07330..0176d3189e 100644
--- a/qpid/extras/dispatch/src/dispatch.c
+++ b/qpid/extras/dispatch/src/dispatch.c
@@ -20,6 +20,7 @@
 #include <qpid/dispatch.h>
 #include "dispatch_private.h"
 #include "alloc_private.h"
+#include "log_private.h"
 
 /**
  * Private Function Prototypes
@@ -42,6 +43,7 @@ dx_dispatch_t *dx_dispatch(int thread_count, const char *container_name,
 {
     dx_dispatch_t *dx = NEW(dx_dispatch_t);
 
+    dx_log_initialize();
     dx_alloc_initialize();
 
     if (!container_name)
@@ -72,5 +74,6 @@ void dx_dispatch_free(dx_dispatch_t *dx)
     dx_router_free(dx->router);
     dx_container_free(dx->container);
     dx_server_free(dx->server);
+    dx_log_finalize();
 }
 
diff --git a/qpid/extras/dispatch/src/log.c b/qpid/extras/dispatch/src/log.c
index c6cffe0321..711d478e32 100644
--- a/qpid/extras/dispatch/src/log.c
+++ b/qpid/extras/dispatch/src/log.c
@@ -17,9 +17,10 @@
  * under the License.
  */
 
-#include <qpid/dispatch/log.h>
+#include "log_private.h"
 #include <qpid/dispatch/ctools.h>
 #include <qpid/dispatch/alloc.h>
+#include <qpid/dispatch/threading.h>
 #include <stdarg.h>
 #include <stdio.h>
 #include <string.h>
@@ -47,7 +48,8 @@ DEQ_DECLARE(dx_log_entry_t, dx_log_list_t);
 
 static int            mask = LOG_INFO;
 static dx_log_list_t  entries;
-static int            list_init = 0;
+static sys_mutex_t   *log_lock = 0;
+
 
 static char *cls_prefix(int cls)
 {
@@ -65,11 +67,6 @@ void dx_log_impl(const char *module, int cls, const char *file, int line, const
     if (!(cls & mask))
         return;
 
-    if (list_init == 0) {
-        list_init = 1;
-        DEQ_INIT(entries);
-    }
-
     dx_log_entry_t *entry = new_dx_log_entry_t();
     entry->module = module;
     entry->cls    = cls;
@@ -84,12 +81,14 @@ void dx_log_impl(const char *module, int cls, const char *file, int line, const
     va_end(ap);
     fprintf(stderr, "%s (%s) %s\n", module, cls_prefix(cls), entry->text);
 
+    sys_mutex_lock(log_lock);
     DEQ_INSERT_TAIL(entries, entry);
     if (DEQ_SIZE(entries) > LIST_MAX) {
         entry = DEQ_HEAD(entries);
         DEQ_REMOVE_HEAD(entries);
         free_dx_log_entry_t(entry);
     }
+    sys_mutex_unlock(log_lock);
 }
 
 void dx_log_set_mask(int _mask)
@@ -97,3 +96,16 @@ void dx_log_set_mask(int _mask)
     mask = _mask;
 }
 
+
+void dx_log_initialize(void)
+{
+    DEQ_INIT(entries);
+    log_lock = sys_mutex();
+}
+
+
+void dx_log_finalize(void)
+{
+}
+
+
diff --git a/qpid/extras/dispatch/src/log_private.h b/qpid/extras/dispatch/src/log_private.h
new file mode 100644
index 0000000000..731822ec6f
--- /dev/null
+++ b/qpid/extras/dispatch/src/log_private.h
@@ -0,0 +1,27 @@
+#ifndef __log_private_h__
+#define __log_private_h__ 1
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <qpid/dispatch/log.h>
+
+void dx_log_initialize(void);
+void dx_log_finalize(void);
+
+#endif
diff --git a/qpid/extras/dispatch/src/message.c b/qpid/extras/dispatch/src/message.c
index fd08753f4d..cf500fdadf 100644
--- a/qpid/extras/dispatch/src/message.c
+++ b/qpid/extras/dispatch/src/message.c
@@ -165,9 +165,9 @@ static int start_list(unsigned char **cursor, dx_buffer_t **buffer)
 //
 static int dx_check_and_advance(dx_buffer_t         **buffer,
                                 unsigned char       **cursor,
-                                unsigned char        *pattern,
+                                const unsigned char  *pattern,
                                 int                   pattern_length,
-                                unsigned char        *expected_tags,
+                                const unsigned char  *expected_tags,
                                 dx_field_location_t  *location)
 {
     dx_buffer_t   *test_buffer = *buffer;
@@ -451,8 +451,9 @@ dx_message_t *dx_allocate_message()
     }
 
     memset(msg->content, 0, sizeof(dx_message_content_t));
-    msg->content->lock      = sys_mutex();
-    msg->content->ref_count = 1;
+    msg->content->lock        = sys_mutex();
+    msg->content->ref_count   = 1;
+    msg->content->parse_depth = DX_DEPTH_NONE;
 
     return (dx_message_t*) msg;
 }
@@ -631,126 +632,146 @@ void dx_message_send(dx_message_t *in_msg, pn_link_t *link)
 }
 
 
-int dx_message_check(dx_message_t *in_msg, dx_message_depth_t depth)
+static int dx_check_field_LH(dx_message_content_t *content,
+                             dx_message_depth_t    depth,
+                             const unsigned char  *long_pattern,
+                             const unsigned char  *short_pattern,
+                             const unsigned char  *expected_tags,
+                             dx_field_location_t  *location)
 {
-
 #define LONG  10
 #define SHORT 3
-#define MSG_HDR_LONG                  (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x70"
-#define MSG_HDR_SHORT                 (unsigned char*) "\x00\x53\x70"
-#define DELIVERY_ANNOTATION_LONG      (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x71"
-#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 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"
-#define BODY_DATA_SHORT               (unsigned char*) "\x00\x53\x75"
-#define BODY_SEQUENCE_LONG            (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x76"
-#define BODY_SEQUENCE_SHORT           (unsigned char*) "\x00\x53\x76"
-#define FOOTER_LONG                   (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x78"
-#define FOOTER_SHORT                  (unsigned char*) "\x00\x53\x78"
-#define TAGS_LIST                     (unsigned char*) "\x45\xc0\xd0"
-#define TAGS_MAP                      (unsigned char*) "\xc1\xd1"
-#define TAGS_BINARY                   (unsigned char*) "\xa0\xb0"
-
-    dx_message_pvt_t     *msg = (dx_message_pvt_t*) in_msg;
-    dx_message_content_t *content = msg->content;
-    dx_buffer_t          *buffer = DEQ_HEAD(content->buffers);
-    unsigned char        *cursor;
+    if (depth > content->parse_depth) {
+        if (0 == dx_check_and_advance(&content->parse_buffer, &content->parse_cursor, long_pattern,  LONG,  expected_tags, location))
+            return 0;
+        if (0 == dx_check_and_advance(&content->parse_buffer, &content->parse_cursor, short_pattern, SHORT, expected_tags, location))
+            return 0;
+        content->parse_depth = depth;
+    }
+    return 1;
+}
+
+
+static int dx_message_check_LH(dx_message_content_t *content, dx_message_depth_t depth)
+{
+    static const unsigned char * const MSG_HDR_LONG                 = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x70";
+    static const unsigned char * const MSG_HDR_SHORT                = (unsigned char*) "\x00\x53\x70";
+    static const unsigned char * const DELIVERY_ANNOTATION_LONG     = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x71";
+    static const unsigned char * const DELIVERY_ANNOTATION_SHORT    = (unsigned char*) "\x00\x53\x71";
+    static const unsigned char * const MESSAGE_ANNOTATION_LONG      = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x72";
+    static const unsigned char * const MESSAGE_ANNOTATION_SHORT     = (unsigned char*) "\x00\x53\x72";
+    static const unsigned char * const PROPERTIES_LONG              = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x73";
+    static const unsigned char * const PROPERTIES_SHORT             = (unsigned char*) "\x00\x53\x73";
+    static const unsigned char * const APPLICATION_PROPERTIES_LONG  = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x74";
+    static const unsigned char * const APPLICATION_PROPERTIES_SHORT = (unsigned char*) "\x00\x53\x74";
+    static const unsigned char * const BODY_DATA_LONG               = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x75";
+    static const unsigned char * const BODY_DATA_SHORT              = (unsigned char*) "\x00\x53\x75";
+    static const unsigned char * const BODY_SEQUENCE_LONG           = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x76";
+    static const unsigned char * const BODY_SEQUENCE_SHORT          = (unsigned char*) "\x00\x53\x76";
+    static const unsigned char * const FOOTER_LONG                  = (unsigned char*) "\x00\x80\x00\x00\x00\x00\x00\x00\x00\x78";
+    static const unsigned char * const FOOTER_SHORT                 = (unsigned char*) "\x00\x53\x78";
+    static const unsigned char * const TAGS_LIST                    = (unsigned char*) "\x45\xc0\xd0";
+    static const unsigned char * const TAGS_MAP                     = (unsigned char*) "\xc1\xd1";
+    static const unsigned char * const TAGS_BINARY                  = (unsigned char*) "\xa0\xb0";
+
+    dx_buffer_t *buffer  = DEQ_HEAD(content->buffers);
 
     if (!buffer)
         return 0; // Invalid - No data in the message
 
+    if (depth <= content->parse_depth)
+        return 1; // We've already parsed at least this deep
+
+    if (content->parse_buffer == 0) {
+        content->parse_buffer = buffer;
+        content->parse_cursor = dx_buffer_base(content->parse_buffer);
+    }
+
     if (depth == DX_DEPTH_NONE)
         return 1;
 
-    cursor = dx_buffer_base(buffer);
-
     //
     // MESSAGE HEADER
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, MSG_HDR_LONG,  LONG,  TAGS_LIST, &content->section_message_header))
-        return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, MSG_HDR_SHORT, SHORT, TAGS_LIST, &content->section_message_header))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_HEADER,
+                               MSG_HDR_LONG, MSG_HDR_SHORT, TAGS_LIST, &content->section_message_header))
         return 0;
-
     if (depth == DX_DEPTH_HEADER)
         return 1;
 
     //
     // DELIVERY ANNOTATION
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_LONG,  LONG,  TAGS_MAP,  &content->section_delivery_annotation))
-        return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_SHORT, SHORT, TAGS_MAP,  &content->section_delivery_annotation))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_DELIVERY_ANNOTATIONS,
+                               DELIVERY_ANNOTATION_LONG, DELIVERY_ANNOTATION_SHORT, TAGS_MAP, &content->section_delivery_annotation))
         return 0;
-
     if (depth == DX_DEPTH_DELIVERY_ANNOTATIONS)
         return 1;
 
     //
     // MESSAGE ANNOTATION
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_LONG,  LONG,  TAGS_MAP,  &content->section_message_annotation))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_MESSAGE_ANNOTATIONS,
+                               MESSAGE_ANNOTATION_LONG, MESSAGE_ANNOTATION_SHORT, TAGS_MAP, &content->section_message_annotation))
         return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_SHORT, SHORT, TAGS_MAP,  &content->section_message_annotation))
-        return 0;
-
     if (depth == DX_DEPTH_MESSAGE_ANNOTATIONS)
         return 1;
 
     //
     // PROPERTIES
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, PROPERTIES_LONG,  LONG,  TAGS_LIST, &content->section_message_properties))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_PROPERTIES,
+                               PROPERTIES_LONG, PROPERTIES_SHORT, TAGS_LIST, &content->section_message_properties))
         return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, PROPERTIES_SHORT, SHORT, TAGS_LIST, &content->section_message_properties))
-        return 0;
-
     if (depth == DX_DEPTH_PROPERTIES)
         return 1;
 
     //
     // APPLICATION PROPERTIES
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_LONG,  LONG,  TAGS_MAP, &content->section_application_properties))
-        return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_SHORT, SHORT, TAGS_MAP, &content->section_application_properties))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_APPLICATION_PROPERTIES,
+                               APPLICATION_PROPERTIES_LONG, APPLICATION_PROPERTIES_SHORT, TAGS_MAP, &content->section_application_properties))
         return 0;
-
     if (depth == DX_DEPTH_APPLICATION_PROPERTIES)
         return 1;
 
     //
     // BODY  (Note that this function expects a single data section or a single AMQP sequence)
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, BODY_DATA_LONG,      LONG,  TAGS_BINARY, &content->section_body))
-        return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, BODY_DATA_SHORT,     SHORT, TAGS_BINARY, &content->section_body))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_BODY,
+                               BODY_DATA_LONG, BODY_DATA_SHORT, TAGS_BINARY, &content->section_body))
         return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_LONG,  LONG,  TAGS_LIST,   &content->section_body))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_BODY,
+                               BODY_SEQUENCE_LONG, BODY_SEQUENCE_SHORT, TAGS_LIST, &content->section_body))
         return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_SHORT, SHORT, TAGS_LIST,   &content->section_body))
-        return 0;
-
     if (depth == DX_DEPTH_BODY)
         return 1;
 
     //
     // FOOTER
     //
-    if (0 == dx_check_and_advance(&buffer, &cursor, FOOTER_LONG,  LONG,  TAGS_MAP, &content->section_footer))
-        return 0;
-    if (0 == dx_check_and_advance(&buffer, &cursor, FOOTER_SHORT, SHORT, TAGS_MAP, &content->section_footer))
+    if (0 == dx_check_field_LH(content, DX_DEPTH_ALL,
+                               FOOTER_LONG, FOOTER_SHORT, TAGS_MAP, &content->section_footer))
         return 0;
 
     return 1;
 }
 
 
+int dx_message_check(dx_message_t *in_msg, dx_message_depth_t depth)
+{
+    dx_message_pvt_t     *msg     = (dx_message_pvt_t*) in_msg;
+    dx_message_content_t *content = msg->content;
+    int                   result;
+
+    sys_mutex_lock(content->lock);
+    result = dx_message_check_LH(content, depth);
+    sys_mutex_unlock(content->lock);
+
+    return result;
+}
+
+
 dx_field_iterator_t *dx_message_field_iterator(dx_message_t *msg, dx_message_field_t field)
 {
     dx_field_location_t *loc = dx_message_field_location(msg, field);
diff --git a/qpid/extras/dispatch/src/message_private.h b/qpid/extras/dispatch/src/message_private.h
index 5fb18078f5..891ceccee3 100644
--- a/qpid/extras/dispatch/src/message_private.h
+++ b/qpid/extras/dispatch/src/message_private.h
@@ -78,6 +78,9 @@ typedef struct {
     dx_field_location_t  compose_count;
     uint32_t             length;
     uint32_t             count;
+    dx_buffer_t         *parse_buffer;
+    unsigned char       *parse_cursor;
+    dx_message_depth_t   parse_depth;
 } dx_message_content_t;
 
 typedef struct {
diff --git a/qpid/extras/dispatch/src/router_node.c b/qpid/extras/dispatch/src/router_node.c
index 65756be215..872b8363c9 100644
--- a/qpid/extras/dispatch/src/router_node.c
+++ b/qpid/extras/dispatch/src/router_node.c
@@ -30,7 +30,7 @@ static char *module = "ROUTER";
 /**
  * Address Types and Processing:
  *
- *   Address                              Hash Compare   onReceive         onEmit
+ *   Address                              Hash Key       onReceive         onEmit
  *   =============================================================================
  *   _local/<local>                       L<local>       handler           forward
  *   _topo/<area>/<router>/<local>        A<area>        forward           forward
@@ -174,6 +174,10 @@ static void router_rx_handler(void* context, dx_link_t *link, pn_delivery_t *del
                 // cases for forwarding.
                 //
                 // Forward to the in-process handler for this message if there is one.
+                // Note: If the handler is going to queue the message for deferred processing,
+                //       it must copy the message.  This function assumes that the handler
+                //       will process the message synchronously and be finished with it upon
+                //       completion.
                 //
                 if (addr->handler)
                     addr->handler(addr->handler_context, msg);
@@ -183,8 +187,9 @@ static void router_rx_handler(void* context, dx_link_t *link, pn_delivery_t *del
                 // TODO - Don't forward if this is a "_local" address.
                 //
                 if (addr->rlink) {
-                    pn_link_t* pn_outlink = dx_link_pn(addr->rlink->link);
-                    DEQ_INSERT_TAIL(addr->rlink->out_fifo, msg);
+                    pn_link_t    *pn_outlink = dx_link_pn(addr->rlink->link);
+                    dx_message_t *copy       = dx_message_copy(msg);
+                    DEQ_INSERT_TAIL(addr->rlink->out_fifo, copy);
                     pn_link_offered(pn_outlink, DEQ_SIZE(addr->rlink->out_fifo));
                     dx_link_activate(addr->rlink->link);
                 }
@@ -205,6 +210,7 @@ static void router_rx_handler(void* context, dx_link_t *link, pn_delivery_t *del
                 pn_delivery_settle(delivery);
             }
 
+            dx_free_message(msg);
             sys_mutex_unlock(router->lock);
         }
     } else {
@@ -286,7 +292,7 @@ static int router_incoming_link_handler(void* context, dx_link_t *link)
 
         pn_terminus_copy(pn_link_source(pn_link), pn_link_remote_source(pn_link));
         pn_terminus_copy(pn_link_target(pn_link), pn_link_remote_target(pn_link));
-        pn_link_flow(pn_link, 8);
+        pn_link_flow(pn_link, 32);
         pn_link_open(pn_link);
     } else {
         pn_link_close(pn_link);
diff --git a/qpid/extras/dispatch/tests/CMakeLists.txt b/qpid/extras/dispatch/tests/CMakeLists.txt
index 362aac08e7..ddf41d0834 100644
--- a/qpid/extras/dispatch/tests/CMakeLists.txt
+++ b/qpid/extras/dispatch/tests/CMakeLists.txt
@@ -22,8 +22,6 @@
 ##
 set(unit_test_SOURCES
     alloc_test.c
-    field_test.c
-    message_test.c
     run_unit_tests.c
     server_test.c
     timer_test.c
@@ -33,7 +31,17 @@ set(unit_test_SOURCES
 add_executable(unit_tests ${unit_test_SOURCES})
 target_link_libraries(unit_tests qpid-dispatch)
 
-add_test(unit_tests_buf_10000 unit_tests 10000)
-add_test(unit_tests_buf_512   unit_tests 512)
-add_test(unit_tests_buf_10    unit_tests 10)
-add_test(unit_tests_buf_1     unit_tests 1)
+set(unit_test_size_SOURCES
+    field_test.c
+    message_test.c
+    run_unit_tests_size.c
+    )
+
+add_executable(unit_tests_size ${unit_test_size_SOURCES})
+target_link_libraries(unit_tests_size qpid-dispatch)
+
+add_test(unit_tests_size_10000 unit_tests_size 10000)
+add_test(unit_tests_size_512   unit_tests_size 512)
+add_test(unit_tests_size_10    unit_tests_size 10)
+add_test(unit_tests_size_1     unit_tests_size 1)
+add_test(unit_tests            unit_tests)
diff --git a/qpid/extras/dispatch/tests/message_test.c b/qpid/extras/dispatch/tests/message_test.c
index ef81dc6d25..2873f48740 100644
--- a/qpid/extras/dispatch/tests/message_test.c
+++ b/qpid/extras/dispatch/tests/message_test.c
@@ -150,6 +150,35 @@ static char* test_insufficient_check_depth(void *context)
 }
 
 
+static char* test_check_multiple(void *context)
+{
+    pn_message_t *pn_msg = pn_message();
+    pn_message_set_address(pn_msg, "test_addr_2");
+
+    size_t size = 10000;
+    int result = pn_message_encode(pn_msg, buffer, &size);
+    if (result != 0) return "Error in pn_message_encode";
+
+    dx_message_t         *msg     = dx_allocate_message();
+    dx_message_content_t *content = MSG_CONTENT(msg);
+
+    set_content(content, size);
+
+    int valid = dx_message_check(msg, DX_DEPTH_DELIVERY_ANNOTATIONS);
+    if (!valid) return "dx_message_check returns 'invalid' for DELIVERY_ANNOTATIONS";
+
+    valid = dx_message_check(msg, DX_DEPTH_BODY);
+    if (!valid) return "dx_message_check returns 'invalid' for BODY";
+
+    valid = dx_message_check(msg, DX_DEPTH_PROPERTIES);
+    if (!valid) return "dx_message_check returns 'invalid' for PROPERTIES";
+
+    dx_free_message(msg);
+
+    return 0;
+}
+
+
 int message_tests(void)
 {
     int result = 0;
@@ -157,6 +186,7 @@ int message_tests(void)
     TEST_CASE(test_send_to_messenger, 0);
     TEST_CASE(test_receive_from_messenger, 0);
     TEST_CASE(test_insufficient_check_depth, 0);
+    TEST_CASE(test_check_multiple, 0);
 
     return result;
 }
diff --git a/qpid/extras/dispatch/tests/run_unit_tests.c b/qpid/extras/dispatch/tests/run_unit_tests.c
index 01a8ae16b3..9b0b5b100e 100644
--- a/qpid/extras/dispatch/tests/run_unit_tests.c
+++ b/qpid/extras/dispatch/tests/run_unit_tests.c
@@ -23,28 +23,14 @@ int tool_tests();
 int timer_tests();
 int alloc_tests();
 int server_tests();
-int message_tests();
-int field_tests();
 
 int main(int argc, char** argv)
 {
-    ssize_t buffer_size = 512;
-
-    if (argc > 1) {
-        buffer_size = atoi(argv[1]);
-        if (buffer_size < 1)
-            return 1;
-    }
-
-    dx_buffer_set_size(buffer_size);
-
     int result = 0;
     result += tool_tests();
     result += timer_tests();
     result += alloc_tests();
     result += server_tests();
-    result += message_tests();
-    result += field_tests();
     return result;
 }
 
diff --git a/qpid/extras/dispatch/tests/run_unit_tests_size.c b/qpid/extras/dispatch/tests/run_unit_tests_size.c
new file mode 100644
index 0000000000..1c6a78c38a
--- /dev/null
+++ b/qpid/extras/dispatch/tests/run_unit_tests_size.c
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <qpid/dispatch/buffer.h>
+#include "alloc_private.h"
+
+int message_tests();
+int field_tests();
+
+int main(int argc, char** argv)
+{
+    ssize_t buffer_size = 512;
+
+    if (argc > 1) {
+        buffer_size = atoi(argv[1]);
+        if (buffer_size < 1)
+            return 1;
+    }
+
+    dx_alloc_initialize();
+    dx_buffer_set_size(buffer_size);
+
+    int result = 0;
+    result += message_tests();
+    result += field_tests();
+    return result;
+}
+