You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2013/01/16 00:28:41 UTC
svn commit: r1433735 [2/3] - in /qpid/trunk/qpid/extras/nexus: ./ include/
include/qpid/ include/qpid/nexus/ site/ site/css/ site/images/
site/includes/ src/ src/posix/ tests/
Added: qpid/trunk/qpid/extras/nexus/src/auth.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/auth.h?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/auth.h (added)
+++ qpid/trunk/qpid/extras/nexus/src/auth.h Tue Jan 15 23:28:40 2013
@@ -0,0 +1,27 @@
+#ifndef __auth_h__
+#define __auth_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 <proton/driver.h>
+
+void auth_client_handler(pn_connector_t *conn);
+void auth_server_handler(pn_connector_t *conn);
+
+#endif
Propchange: qpid/trunk/qpid/extras/nexus/src/auth.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/extras/nexus/src/auth.h
------------------------------------------------------------------------------
svn:keywords = Author Date Id Rev URL
Added: qpid/trunk/qpid/extras/nexus/src/container.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/container.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/container.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/container.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,620 @@
+/*
+ * 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 <stdio.h>
+#include <string.h>
+#include <qpid/nexus/container.h>
+#include <qpid/nexus/message.h>
+#include <proton/engine.h>
+#include <proton/message.h>
+#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/hash.h>
+#include <qpid/nexus/threading.h>
+#include <qpid/nexus/iterator.h>
+#include <qpid/nexus/log.h>
+
+static char *module="CONTAINER";
+
+struct nx_node_t {
+ const nx_node_type_t *ntype;
+ char *name;
+ void *context;
+ nx_dist_mode_t supported_dist;
+ nx_lifetime_policy_t life_policy;
+};
+
+ALLOC_DECLARE(nx_node_t);
+ALLOC_DEFINE(nx_node_t);
+ALLOC_DEFINE(nx_link_item_t);
+
+struct nx_link_t {
+ pn_link_t *pn_link;
+ void *context;
+ nx_node_t *node;
+};
+
+ALLOC_DECLARE(nx_link_t);
+ALLOC_DEFINE(nx_link_t);
+
+typedef struct nxc_node_type_t {
+ DEQ_LINKS(struct nxc_node_type_t);
+ const nx_node_type_t *ntype;
+} nxc_node_type_t;
+DEQ_DECLARE(nxc_node_type_t, nxc_node_type_list_t);
+
+
+static hash_t *node_type_map;
+static hash_t *node_map;
+static sys_mutex_t *lock;
+static nx_node_t *default_node;
+static nxc_node_type_list_t node_type_list;
+
+static void setup_outgoing_link(pn_link_t *pn_link)
+{
+ sys_mutex_lock(lock);
+ nx_node_t *node;
+ int result;
+ const char *source = pn_terminus_get_address(pn_link_remote_source(pn_link));
+ nx_field_iterator_t *iter;
+ // TODO - Extract the name from the structured source
+
+ if (source) {
+ iter = nx_field_iterator_string(source, ITER_VIEW_NODE_ID);
+ result = hash_retrieve(node_map, iter, (void*) &node);
+ nx_field_iterator_free(iter);
+ } else
+ result = -1;
+ sys_mutex_unlock(lock);
+
+ if (result < 0) {
+ if (default_node)
+ node = default_node;
+ else {
+ // Reject the link
+ // TODO - When the API allows, add an error message for "no available node"
+ pn_link_close(pn_link);
+ return;
+ }
+ }
+
+ nx_link_t *link = new_nx_link_t();
+ if (!link) {
+ pn_link_close(pn_link);
+ return;
+ }
+
+ link->pn_link = pn_link;
+ link->context = 0;
+ link->node = node;
+
+ pn_link_set_context(pn_link, link);
+ node->ntype->outgoing_handler(node->context, link);
+}
+
+
+static void setup_incoming_link(pn_link_t *pn_link)
+{
+ sys_mutex_lock(lock);
+ nx_node_t *node;
+ int result;
+ const char *target = pn_terminus_get_address(pn_link_remote_target(pn_link));
+ nx_field_iterator_t *iter;
+ // TODO - Extract the name from the structured target
+
+ if (target) {
+ iter = nx_field_iterator_string(target, ITER_VIEW_NODE_ID);
+ result = hash_retrieve(node_map, iter, (void*) &node);
+ nx_field_iterator_free(iter);
+ } else
+ result = -1;
+ sys_mutex_unlock(lock);
+
+ if (result < 0) {
+ if (default_node)
+ node = default_node;
+ else {
+ // Reject the link
+ // TODO - When the API allows, add an error message for "no available node"
+ pn_link_close(pn_link);
+ return;
+ }
+ }
+
+ nx_link_t *link = new_nx_link_t();
+ if (!link) {
+ pn_link_close(pn_link);
+ return;
+ }
+
+ link->pn_link = pn_link;
+ link->context = 0;
+ link->node = node;
+
+ pn_link_set_context(pn_link, link);
+ node->ntype->incoming_handler(node->context, link);
+}
+
+
+static int do_writable(pn_link_t *pn_link)
+{
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+ if (!link)
+ return 0;
+
+ nx_node_t *node = link->node;
+ if (!node)
+ return 0;
+
+ return node->ntype->writable_handler(node->context, link);
+}
+
+
+static void process_receive(pn_delivery_t *delivery)
+{
+ pn_link_t *pn_link = pn_delivery_link(delivery);
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+
+ if (link) {
+ nx_node_t *node = link->node;
+ if (node) {
+ node->ntype->rx_handler(node->context, link, delivery);
+ return;
+ }
+ }
+
+ //
+ // Reject the delivery if we couldn't find a node to handle it
+ //
+ pn_link_advance(pn_link);
+ pn_link_flow(pn_link, 1);
+ pn_delivery_update(delivery, PN_REJECTED);
+ pn_delivery_settle(delivery);
+}
+
+
+static void do_send(pn_delivery_t *delivery)
+{
+ pn_link_t *pn_link = pn_delivery_link(delivery);
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+
+ if (link) {
+ nx_node_t *node = link->node;
+ if (node) {
+ node->ntype->tx_handler(node->context, link, delivery);
+ return;
+ }
+ }
+
+ // TODO - Cancel the delivery
+}
+
+
+static void do_updated(pn_delivery_t *delivery)
+{
+ pn_link_t *pn_link = pn_delivery_link(delivery);
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+
+ if (link) {
+ nx_node_t *node = link->node;
+ if (node)
+ node->ntype->disp_handler(node->context, link, delivery);
+ }
+}
+
+
+static int close_handler(void* unused, pn_connection_t *conn)
+{
+ //
+ // Close all links, passing False as the 'closed' argument. These links are not
+ // being properly 'detached'. They are being orphaned.
+ //
+ pn_link_t *pn_link = pn_link_head(conn, 0);
+ while (pn_link) {
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+ nx_node_t *node = link->node;
+ if (node)
+ node->ntype->link_detach_handler(node->context, link, 0);
+ pn_link_close(pn_link);
+ free_nx_link_t(link);
+ pn_link = pn_link_next(pn_link, 0);
+ }
+
+ // teardown all sessions
+ pn_session_t *ssn = pn_session_head(conn, 0);
+ while (ssn) {
+ pn_session_close(ssn);
+ ssn = pn_session_next(ssn, 0);
+ }
+
+ // teardown the connection
+ pn_connection_close(conn);
+ return 0;
+}
+
+
+static int process_handler(void* unused, pn_connection_t *conn)
+{
+ pn_session_t *ssn;
+ pn_link_t *pn_link;
+ pn_delivery_t *delivery;
+ int event_count = 0;
+
+ // Step 1: setup the engine's connection, and any sessions and links
+ // that may be pending.
+
+ // initialize the connection if it's new
+ if (pn_connection_state(conn) & PN_LOCAL_UNINIT) {
+ pn_connection_open(conn);
+ event_count++;
+ }
+
+ // open all pending sessions
+ ssn = pn_session_head(conn, PN_LOCAL_UNINIT);
+ while (ssn) {
+ pn_session_open(ssn);
+ ssn = pn_session_next(ssn, PN_LOCAL_UNINIT);
+ event_count++;
+ }
+
+ // configure and open any pending links
+ pn_link = pn_link_head(conn, PN_LOCAL_UNINIT);
+ while (pn_link) {
+ if (pn_link_is_sender(pn_link))
+ setup_outgoing_link(pn_link);
+ else
+ setup_incoming_link(pn_link);
+ pn_link = pn_link_next(pn_link, PN_LOCAL_UNINIT);
+ event_count++;
+ }
+
+
+ // Step 2: Now drain all the pending deliveries from the connection's
+ // work queue and process them
+
+ delivery = pn_work_head(conn);
+ while (delivery) {
+ if (pn_delivery_readable(delivery))
+ process_receive(delivery);
+ else if (pn_delivery_writable(delivery))
+ do_send(delivery);
+
+ if (pn_delivery_updated(delivery))
+ do_updated(delivery);
+
+ delivery = pn_work_next(delivery);
+ event_count++;
+ }
+
+ //
+ // Step 2.5: Traverse all of the links on the connection looking for
+ // outgoing links with non-zero credit. Call the attached node's
+ // writable handler for such links.
+ //
+ pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE);
+ while (pn_link) {
+ assert(pn_session_connection(pn_link_session(pn_link)) == conn);
+ if (pn_link_is_sender(pn_link) && pn_link_credit(pn_link) > 0)
+ event_count += do_writable(pn_link);
+ pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE);
+ }
+
+ // Step 3: Clean up any links or sessions that have been closed by the
+ // remote. If the connection has been closed remotely, clean that up
+ // also.
+
+ // teardown any terminating links
+ pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
+ while (pn_link) {
+ nx_link_t *link = (nx_link_t*) pn_link_get_context(pn_link);
+ nx_node_t *node = link->node;
+ if (node)
+ node->ntype->link_detach_handler(node->context, link, 1); // TODO - get 'closed' from detach message
+ pn_link_close(pn_link);
+ pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
+ event_count++;
+ }
+
+ // teardown any terminating sessions
+ ssn = pn_session_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
+ while (ssn) {
+ pn_session_close(ssn);
+ ssn = pn_session_next(ssn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
+ event_count++;
+ }
+
+ // teardown the connection if it's terminating
+ if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
+ pn_connection_close(conn);
+ event_count++;
+ }
+
+ return event_count;
+}
+
+
+static void open_handler(nx_connection_t *conn, nx_direction_t dir)
+{
+ const nx_node_type_t *nt;
+
+ //
+ // Note the locking structure in this function. Generally this would be unsafe, but since
+ // this particular list is only ever appended to and never has items inserted or deleted,
+ // this usage is safe in this case.
+ //
+ sys_mutex_lock(lock);
+ nxc_node_type_t *nt_item = DEQ_HEAD(node_type_list);
+ sys_mutex_unlock(lock);
+
+ pn_connection_open(nx_connection_pn(conn));
+
+ while (nt_item) {
+ nt = nt_item->ntype;
+ if (dir == NX_INCOMING) {
+ if (nt->inbound_conn_open_handler)
+ nt->inbound_conn_open_handler(nt->type_context, conn);
+ } else {
+ if (nt->outbound_conn_open_handler)
+ nt->outbound_conn_open_handler(nt->type_context, conn);
+ }
+
+ sys_mutex_lock(lock);
+ nt_item = DEQ_NEXT(nt_item);
+ sys_mutex_unlock(lock);
+ }
+}
+
+
+static int handler(void* context, nx_conn_event_t event, nx_connection_t *nx_conn)
+{
+ pn_connection_t *conn = nx_connection_pn(nx_conn);
+
+ switch (event) {
+ case NX_CONN_EVENT_LISTENER_OPEN: open_handler(nx_conn, NX_INCOMING); break;
+ case NX_CONN_EVENT_CONNECTOR_OPEN: open_handler(nx_conn, NX_OUTGOING); break;
+ case NX_CONN_EVENT_CLOSE: return close_handler(context, conn);
+ case NX_CONN_EVENT_PROCESS: return process_handler(context, conn);
+ }
+
+ return 0;
+}
+
+
+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();
+ default_node = 0;
+ DEQ_INIT(node_type_list);
+
+ nx_server_set_conn_handler(handler);
+}
+
+
+void nx_container_finalize(void)
+{
+}
+
+
+int nx_container_register_node_type(const nx_node_type_t *nt)
+{
+ int result;
+ nx_field_iterator_t *iter = nx_field_iterator_string(nt->type_name, ITER_VIEW_ALL);
+ nxc_node_type_t *nt_item = NEW(nxc_node_type_t);
+ DEQ_ITEM_INIT(nt_item);
+ nt_item->ntype = nt;
+
+ sys_mutex_lock(lock);
+ result = hash_insert_const(node_type_map, iter, nt);
+ DEQ_INSERT_TAIL(node_type_list, nt_item);
+ sys_mutex_unlock(lock);
+
+ nx_field_iterator_free(iter);
+ if (result < 0)
+ return result;
+ nx_log(module, LOG_TRACE, "Node Type Registered - %s", nt->type_name);
+
+ return 0;
+}
+
+
+void nx_container_set_default_node_type(const nx_node_type_t *nt,
+ void *context,
+ nx_dist_mode_t supported_dist)
+{
+ if (default_node)
+ nx_container_destroy_node(default_node);
+
+ if (nt) {
+ default_node = nx_container_create_node(nt, 0, context, supported_dist, NX_LIFE_PERMANENT);
+ nx_log(module, LOG_TRACE, "Node of type '%s' installed as default node", nt->type_name);
+ } else {
+ default_node = 0;
+ nx_log(module, LOG_TRACE, "Default node removed");
+ }
+}
+
+
+nx_node_t *nx_container_create_node(const nx_node_type_t *nt,
+ const char *name,
+ void *context,
+ nx_dist_mode_t supported_dist,
+ nx_lifetime_policy_t life_policy)
+{
+ int result;
+ nx_node_t *node = new_nx_node_t();
+ if (!node)
+ return 0;
+
+ node->ntype = nt;
+ node->name = 0;
+ node->context = context;
+ node->supported_dist = supported_dist;
+ node->life_policy = life_policy;
+
+ if (name) {
+ nx_field_iterator_t *iter = nx_field_iterator_string(name, ITER_VIEW_ALL);
+ sys_mutex_lock(lock);
+ result = hash_insert(node_map, iter, node);
+ sys_mutex_unlock(lock);
+ nx_field_iterator_free(iter);
+ if (result < 0) {
+ free_nx_node_t(node);
+ return 0;
+ }
+
+ node->name = (char*) malloc(strlen(name) + 1);
+ strcpy(node->name, name);
+ }
+
+ if (name)
+ nx_log(module, LOG_TRACE, "Node of type '%s' created with name '%s'", nt->type_name, name);
+
+ return node;
+}
+
+
+void nx_container_destroy_node(nx_node_t *node)
+{
+ if (node->name) {
+ nx_field_iterator_t *iter = nx_field_iterator_string(node->name, ITER_VIEW_ALL);
+ sys_mutex_lock(lock);
+ hash_remove(node_map, iter);
+ sys_mutex_unlock(lock);
+ nx_field_iterator_free(iter);
+ free(node->name);
+ }
+
+ free_nx_node_t(node);
+}
+
+
+void nx_container_node_set_context(nx_node_t *node, void *node_context)
+{
+ node->context = node_context;
+}
+
+
+nx_dist_mode_t nx_container_node_get_dist_modes(const nx_node_t *node)
+{
+ return node->supported_dist;
+}
+
+
+nx_lifetime_policy_t nx_container_node_get_life_policy(const nx_node_t *node)
+{
+ return node->life_policy;
+}
+
+
+nx_link_t *nx_link(nx_node_t *node, nx_connection_t *conn, nx_direction_t dir, const char* name)
+{
+ pn_session_t *sess = pn_session(nx_connection_pn(conn));
+ nx_link_t *link = new_nx_link_t();
+
+ if (dir == NX_OUTGOING)
+ link->pn_link = pn_sender(sess, name);
+ else
+ link->pn_link = pn_receiver(sess, name);
+ link->context = node->context;
+ link->node = node;
+
+ pn_link_set_context(link->pn_link, link);
+
+ pn_session_open(sess);
+
+ return link;
+}
+
+
+void nx_link_set_context(nx_link_t *link, void *context)
+{
+ link->context = context;
+}
+
+
+void *nx_link_get_context(nx_link_t *link)
+{
+ return link->context;
+}
+
+
+pn_link_t *nx_link_pn(nx_link_t *link)
+{
+ return link->pn_link;
+}
+
+
+pn_terminus_t *nx_link_source(nx_link_t *link)
+{
+ return pn_link_source(link->pn_link);
+}
+
+
+pn_terminus_t *nx_link_target(nx_link_t *link)
+{
+ return pn_link_target(link->pn_link);
+}
+
+
+pn_terminus_t *nx_link_remote_source(nx_link_t *link)
+{
+ return pn_link_remote_source(link->pn_link);
+}
+
+
+pn_terminus_t *nx_link_remote_target(nx_link_t *link)
+{
+ return pn_link_remote_target(link->pn_link);
+}
+
+
+void nx_link_activate(nx_link_t *link)
+{
+ if (!link || !link->pn_link)
+ return;
+
+ pn_session_t *sess = pn_link_session(link->pn_link);
+ if (!sess)
+ return;
+
+ pn_connection_t *conn = pn_session_connection(sess);
+ if (!conn)
+ return;
+
+ nx_connection_t *ctx = pn_connection_get_context(conn);
+ if (!ctx)
+ return;
+
+ nx_server_activate(ctx);
+}
+
+
+void nx_link_close(nx_link_t *link)
+{
+ pn_link_close(link->pn_link);
+}
+
+
Propchange: qpid/trunk/qpid/extras/nexus/src/container.c
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/extras/nexus/src/hash.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/hash.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/hash.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/hash.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,223 @@
+/*
+ * 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/nexus/hash.h>
+#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/alloc.h>
+#include <stdio.h>
+#include <string.h>
+
+typedef struct hash_item_t {
+ DEQ_LINKS(struct hash_item_t);
+ unsigned char *key;
+ union {
+ void *val;
+ const void *val_const;
+ } v;
+} hash_item_t;
+
+ALLOC_DECLARE(hash_item_t);
+ALLOC_DEFINE(hash_item_t);
+DEQ_DECLARE(hash_item_t, items_t);
+
+
+typedef struct bucket_t {
+ items_t items;
+} bucket_t;
+
+
+struct hash_t {
+ bucket_t *buckets;
+ unsigned int bucket_count;
+ unsigned int bucket_mask;
+ int batch_size;
+ size_t size;
+ int is_const;
+};
+
+
+// djb2 hash algorithm
+static unsigned long hash_function(nx_field_iterator_t *iter)
+{
+ unsigned long hash = 5381;
+ int c;
+
+ while (!nx_field_iterator_end(iter)) {
+ c = (int) nx_field_iterator_octet(iter);
+ hash = ((hash << 5) + hash) + c; /* hash * 33 + c */
+ }
+
+ return hash;
+}
+
+
+hash_t *hash(int bucket_exponent, int batch_size, int value_is_const)
+{
+ int i;
+ hash_t *h = NEW(hash_t);
+
+ if (!h)
+ return 0;
+
+ h->bucket_count = 1 << bucket_exponent;
+ h->bucket_mask = h->bucket_count - 1;
+ h->batch_size = batch_size;
+ h->size = 0;
+ h->is_const = value_is_const;
+ h->buckets = NEW_ARRAY(bucket_t, h->bucket_count);
+ for (i = 0; i < h->bucket_count; i++) {
+ DEQ_INIT(h->buckets[i].items);
+ }
+
+ return h;
+}
+
+
+void hash_free(hash_t *h)
+{
+ // TODO - Implement this
+}
+
+
+size_t hash_size(hash_t *h)
+{
+ return h ? h->size : 0;
+}
+
+
+static hash_item_t *hash_internal_insert(hash_t *h, nx_field_iterator_t *key, int *error)
+{
+ unsigned long idx = hash_function(key) & h->bucket_mask;
+ hash_item_t *item = DEQ_HEAD(h->buckets[idx].items);
+
+ *error = 0;
+
+ while (item) {
+ if (nx_field_iterator_equal(key, item->key))
+ break;
+ item = item->next;
+ }
+
+ if (item) {
+ *error = -1;
+ return 0;
+ }
+
+ item = new_hash_item_t();
+ if (!item) {
+ *error = -2;
+ return 0;
+ }
+
+ DEQ_ITEM_INIT(item);
+ item->key = nx_field_iterator_copy(key);
+
+ DEQ_INSERT_TAIL(h->buckets[idx].items, item);
+ h->size++;
+ return item;
+}
+
+
+int hash_insert(hash_t *h, nx_field_iterator_t *key, void *val)
+{
+ int error = 0;
+ hash_item_t *item = hash_internal_insert(h, key, &error);
+
+ if (item)
+ item->v.val = val;
+ return error;
+}
+
+
+int hash_insert_const(hash_t *h, nx_field_iterator_t *key, const void *val)
+{
+ if (!h->is_const)
+ return -3;
+
+ int error = 0;
+ hash_item_t *item = hash_internal_insert(h, key, &error);
+
+ if (item)
+ item->v.val_const = val;
+ return error;
+}
+
+
+static hash_item_t *hash_internal_retrieve(hash_t *h, nx_field_iterator_t *key)
+{
+ unsigned long idx = hash_function(key) & h->bucket_mask;
+ hash_item_t *item = DEQ_HEAD(h->buckets[idx].items);
+
+ while (item) {
+ if (nx_field_iterator_equal(key, item->key))
+ break;
+ item = item->next;
+ }
+
+ return item;
+}
+
+
+int hash_retrieve(hash_t *h, nx_field_iterator_t *key, void **val)
+{
+ hash_item_t *item = hash_internal_retrieve(h, key);
+ if (item) {
+ *val = item->v.val;
+ return 0;
+ }
+ return -1;
+}
+
+
+int hash_retrieve_const(hash_t *h, nx_field_iterator_t *key, const void **val)
+{
+ if (!h->is_const)
+ return -3;
+
+ hash_item_t *item = hash_internal_retrieve(h, key);
+ if (item) {
+ *val = item->v.val_const;
+ return 0;
+ }
+ return -1;
+}
+
+
+int hash_remove(hash_t *h, nx_field_iterator_t *key)
+{
+ unsigned long idx = hash_function(key) & h->bucket_mask;
+ hash_item_t *item = DEQ_HEAD(h->buckets[idx].items);
+
+ while (item) {
+ if (nx_field_iterator_equal(key, item->key))
+ break;
+ item = item->next;
+ }
+
+ if (item) {
+ free(item->key);
+ DEQ_REMOVE(h->buckets[idx].items, item);
+ free_hash_item_t(item);
+ h->size--;
+ return 0;
+ }
+
+ return -1;
+}
+
Propchange: qpid/trunk/qpid/extras/nexus/src/hash.c
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/extras/nexus/src/iterator.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/iterator.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/iterator.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/iterator.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,268 @@
+/*
+ * 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/nexus/iterator.h>
+#include <qpid/nexus/message.h>
+#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/alloc.h>
+#include <stdio.h>
+#include <string.h>
+
+typedef enum {
+MODE_TO_END,
+MODE_TO_SLASH
+} parse_mode_t;
+
+struct nx_field_iterator_t {
+ nx_buffer_t *start_buffer;
+ unsigned char *start_cursor;
+ int start_length;
+ nx_buffer_t *buffer;
+ unsigned char *cursor;
+ int length;
+ nx_iterator_view_t view;
+ parse_mode_t mode;
+};
+
+
+ALLOC_DECLARE(nx_field_iterator_t);
+ALLOC_DEFINE(nx_field_iterator_t);
+
+
+typedef enum {
+STATE_START,
+STATE_SLASH_LEFT,
+STATE_SKIPPING_TO_NEXT_SLASH,
+STATE_SCANNING,
+STATE_COLON,
+STATE_COLON_SLASH,
+STATE_AT_NODE_ID
+} state_t;
+
+
+static void view_initialize(nx_field_iterator_t *iter)
+{
+ if (iter->view == ITER_VIEW_ALL) {
+ iter->mode = MODE_TO_END;
+ return;
+ }
+
+ //
+ // Advance to the node-id.
+ //
+ state_t state = STATE_START;
+ unsigned int octet;
+ while (!nx_field_iterator_end(iter) && state != STATE_AT_NODE_ID) {
+ octet = nx_field_iterator_octet(iter);
+ switch (state) {
+ case STATE_START :
+ if (octet == '/')
+ state = STATE_SLASH_LEFT;
+ else
+ state = STATE_SCANNING;
+ break;
+
+ case STATE_SLASH_LEFT :
+ if (octet == '/')
+ state = STATE_SKIPPING_TO_NEXT_SLASH;
+ else
+ state = STATE_AT_NODE_ID;
+ break;
+
+ case STATE_SKIPPING_TO_NEXT_SLASH :
+ if (octet == '/')
+ state = STATE_AT_NODE_ID;
+ break;
+
+ case STATE_SCANNING :
+ if (octet == ':')
+ state = STATE_COLON;
+ break;
+
+ case STATE_COLON :
+ if (octet == '/')
+ state = STATE_COLON_SLASH;
+ else
+ state = STATE_SCANNING;
+ break;
+
+ case STATE_COLON_SLASH :
+ if (octet == '/')
+ state = STATE_SKIPPING_TO_NEXT_SLASH;
+ else
+ state = STATE_SCANNING;
+ break;
+
+ case STATE_AT_NODE_ID :
+ break;
+ }
+ }
+
+ if (state != STATE_AT_NODE_ID) {
+ //
+ // The address string was relative, not absolute. The node-id
+ // is at the beginning of the string.
+ //
+ iter->buffer = iter->start_buffer;
+ iter->cursor = iter->start_cursor;
+ iter->length = iter->start_length;
+ }
+
+ //
+ // Cursor is now on the first octet of the node-id
+ //
+ if (iter->view == ITER_VIEW_NODE_ID) {
+ iter->mode = MODE_TO_SLASH;
+ return;
+ }
+
+ if (iter->view == ITER_VIEW_NO_HOST) {
+ iter->mode = MODE_TO_END;
+ return;
+ }
+
+ if (iter->view == ITER_VIEW_NODE_SPECIFIC) {
+ iter->mode = MODE_TO_END;
+ while (!nx_field_iterator_end(iter)) {
+ octet = nx_field_iterator_octet(iter);
+ if (octet == '/')
+ break;
+ }
+ return;
+ }
+}
+
+
+nx_field_iterator_t* nx_field_iterator_string(const char *text, nx_iterator_view_t view)
+{
+ nx_field_iterator_t *iter = new_nx_field_iterator_t();
+ if (!iter)
+ return 0;
+
+ iter->start_buffer = 0;
+ iter->start_cursor = (unsigned char*) text;
+ iter->start_length = strlen(text);
+
+ nx_field_iterator_reset(iter, view);
+
+ return iter;
+}
+
+
+nx_field_iterator_t *nx_field_iterator_buffer(nx_buffer_t *buffer, int offset, int length, nx_iterator_view_t view)
+{
+ nx_field_iterator_t *iter = new_nx_field_iterator_t();
+ if (!iter)
+ return 0;
+
+ iter->start_buffer = buffer;
+ iter->start_cursor = nx_buffer_base(buffer) + offset;
+ iter->start_length = length;
+
+ nx_field_iterator_reset(iter, view);
+
+ return iter;
+}
+
+
+void nx_field_iterator_free(nx_field_iterator_t *iter)
+{
+ free_nx_field_iterator_t(iter);
+}
+
+
+void nx_field_iterator_reset(nx_field_iterator_t *iter, nx_iterator_view_t view)
+{
+ iter->buffer = iter->start_buffer;
+ iter->cursor = iter->start_cursor;
+ iter->length = iter->start_length;
+ iter->view = view;
+
+ view_initialize(iter);
+}
+
+
+unsigned char nx_field_iterator_octet(nx_field_iterator_t *iter)
+{
+ if (iter->length == 0)
+ return (unsigned char) 0;
+
+ unsigned char result = *(iter->cursor);
+
+ iter->cursor++;
+ iter->length--;
+
+ if (iter->length > 0) {
+ if (iter->buffer) {
+ if (iter->cursor - nx_buffer_base(iter->buffer) == nx_buffer_size(iter->buffer)) {
+ iter->buffer = iter->buffer->next;
+ if (iter->buffer == 0)
+ iter->length = 0;
+ iter->cursor = nx_buffer_base(iter->buffer);
+ }
+ }
+ }
+
+ if (iter->length && iter->mode == MODE_TO_SLASH && *(iter->cursor) == '/')
+ iter->length = 0;
+
+ return result;
+}
+
+
+int nx_field_iterator_end(nx_field_iterator_t *iter)
+{
+ return iter->length == 0;
+}
+
+
+int nx_field_iterator_equal(nx_field_iterator_t *iter, unsigned char *string)
+{
+ nx_field_iterator_reset(iter, iter->view);
+ while (!nx_field_iterator_end(iter) && *string) {
+ if (*string != nx_field_iterator_octet(iter))
+ return 0;
+ string++;
+ }
+
+ return (nx_field_iterator_end(iter) && (*string == 0));
+}
+
+
+unsigned char *nx_field_iterator_copy(nx_field_iterator_t *iter)
+{
+ int length = 0;
+ int idx = 0;
+ unsigned char *copy;
+
+ nx_field_iterator_reset(iter, iter->view);
+ while (!nx_field_iterator_end(iter)) {
+ nx_field_iterator_octet(iter);
+ length++;
+ }
+
+ nx_field_iterator_reset(iter, iter->view);
+ copy = (unsigned char*) malloc(length + 1);
+ while (!nx_field_iterator_end(iter))
+ copy[idx++] = nx_field_iterator_octet(iter);
+ copy[idx] = '\0';
+
+ return copy;
+}
+
Propchange: qpid/trunk/qpid/extras/nexus/src/iterator.c
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/extras/nexus/src/log.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/log.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/log.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/log.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,56 @@
+/*
+ * 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/nexus/log.h>
+#include <stdarg.h>
+#include <stdio.h>
+#include <string.h>
+
+static int mask=LOG_INFO;
+
+static char *cls_prefix(int cls)
+{
+ switch (cls) {
+ case LOG_TRACE : return "TRACE";
+ case LOG_ERROR : return "ERROR";
+ case LOG_INFO : return "INFO";
+ }
+
+ return "";
+}
+
+void nx_log(const char *module, int cls, const char *fmt, ...)
+{
+ if (!(cls & mask))
+ return;
+
+ va_list ap;
+ char line[128];
+
+ va_start(ap, fmt);
+ vsnprintf(line, 127, fmt, ap);
+ va_end(ap);
+ fprintf(stderr, "%s (%s): %s\n", module, cls_prefix(cls), line);
+}
+
+void nx_log_set_mask(int _mask)
+{
+ mask = _mask;
+}
+
Propchange: qpid/trunk/qpid/extras/nexus/src/log.c
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/extras/nexus/src/message.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/message.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/message.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/message.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,1164 @@
+/*
+ * 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/nexus/message.h>
+#include <qpid/nexus/ctools.h>
+#include <qpid/nexus/threading.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;
+}
+
+
+static void advance(unsigned char **cursor, nx_buffer_t **buffer, int consume)
+{
+ unsigned char *local_cursor = *cursor;
+ nx_buffer_t *local_buffer = *buffer;
+
+ int remaining = nx_buffer_size(local_buffer) - (local_cursor - nx_buffer_base(local_buffer));
+ while (consume > 0) {
+ if (consume < remaining) {
+ local_cursor += consume;
+ consume = 0;
+ } else {
+ consume -= remaining;
+ local_buffer = local_buffer->next;
+ if (local_buffer == 0){
+ local_cursor = 0;
+ break;
+ }
+ local_cursor = nx_buffer_base(local_buffer);
+ remaining = nx_buffer_size(local_buffer) - (local_cursor - nx_buffer_base(local_buffer));
+ }
+ }
+
+ *cursor = local_cursor;
+ *buffer = local_buffer;
+}
+
+
+static unsigned char next_octet(unsigned char **cursor, nx_buffer_t **buffer)
+{
+ unsigned char result = **cursor;
+ advance(cursor, buffer, 1);
+ return result;
+}
+
+
+static int traverse_field(unsigned char **cursor, nx_buffer_t **buffer, nx_field_location_t *field)
+{
+ unsigned char tag = next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+ int consume = 0;
+ switch (tag & 0xF0) {
+ case 0x40 : consume = 0; break;
+ case 0x50 : consume = 1; break;
+ case 0x60 : consume = 2; break;
+ case 0x70 : consume = 4; break;
+ case 0x80 : consume = 8; break;
+ case 0x90 : consume = 16; break;
+
+ case 0xB0 :
+ case 0xD0 :
+ case 0xF0 :
+ consume |= ((int) next_octet(cursor, buffer)) << 24;
+ if (!(*cursor)) return 0;
+ consume |= ((int) next_octet(cursor, buffer)) << 16;
+ if (!(*cursor)) return 0;
+ consume |= ((int) next_octet(cursor, buffer)) << 8;
+ if (!(*cursor)) return 0;
+ // Fall through to the next case...
+
+ case 0xA0 :
+ case 0xC0 :
+ case 0xE0 :
+ consume |= (int) next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+ break;
+ }
+
+ if (field) {
+ field->buffer = *buffer;
+ field->offset = *cursor - nx_buffer_base(*buffer);
+ field->length = consume;
+ field->parsed = 1;
+ }
+
+ advance(cursor, buffer, consume);
+ return 1;
+}
+
+
+static int start_list(unsigned char **cursor, nx_buffer_t **buffer)
+{
+ unsigned char tag = next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+ int length = 0;
+ int count = 0;
+
+ switch (tag) {
+ case 0x45 : // list0
+ break;
+ case 0xd0 : // list32
+ length |= ((int) next_octet(cursor, buffer)) << 24;
+ if (!(*cursor)) return 0;
+ length |= ((int) next_octet(cursor, buffer)) << 16;
+ if (!(*cursor)) return 0;
+ length |= ((int) next_octet(cursor, buffer)) << 8;
+ if (!(*cursor)) return 0;
+ length |= (int) next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+
+ count |= ((int) next_octet(cursor, buffer)) << 24;
+ if (!(*cursor)) return 0;
+ count |= ((int) next_octet(cursor, buffer)) << 16;
+ if (!(*cursor)) return 0;
+ count |= ((int) next_octet(cursor, buffer)) << 8;
+ if (!(*cursor)) return 0;
+ count |= (int) next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+
+ break;
+
+ case 0xc0 : // list8
+ length |= (int) next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+
+ count |= (int) next_octet(cursor, buffer);
+ if (!(*cursor)) return 0;
+ break;
+ }
+
+ return count;
+}
+
+
+//
+// Check the buffer chain, starting at cursor to see if it matches the pattern.
+// If the pattern matches, check the next tag to see if it's in the set of expected
+// tags. If not, return zero. If so, set the location descriptor to the good
+// tag and advance the cursor (and buffer, if needed) to the end of the matched section.
+//
+// If there is no match, don't advance the cursor.
+//
+// Return 0 if the pattern matches but the following tag is unexpected
+// Return 0 if the pattern matches and the location already has a pointer (duplicate section)
+// Return 1 if the pattern matches and we've advanced the cursor/buffer
+// Return 1 if the pattern does not match
+//
+static int nx_check_and_advance(nx_buffer_t **buffer,
+ unsigned char **cursor,
+ unsigned char *pattern,
+ int pattern_length,
+ unsigned char *expected_tags,
+ nx_field_location_t *location)
+{
+ nx_buffer_t *test_buffer = *buffer;
+ unsigned char *test_cursor = *cursor;
+
+ if (!test_cursor)
+ return 1; // no match
+
+ unsigned char *end_of_buffer = nx_buffer_base(test_buffer) + nx_buffer_size(test_buffer);
+ int idx = 0;
+
+ while (idx < pattern_length && *test_cursor == pattern[idx]) {
+ idx++;
+ test_cursor++;
+ if (test_cursor == end_of_buffer) {
+ test_buffer = test_buffer->next;
+ if (test_buffer == 0)
+ return 1; // Pattern didn't match
+ test_cursor = nx_buffer_base(test_buffer);
+ end_of_buffer = test_cursor + nx_buffer_size(test_buffer);
+ }
+ }
+
+ if (idx < pattern_length)
+ return 1; // Pattern didn't match
+
+ //
+ // Pattern matched, check the tag
+ //
+ while (*expected_tags && *test_cursor != *expected_tags)
+ expected_tags++;
+ if (*expected_tags == 0)
+ return 0; // Unexpected tag
+
+ if (location->parsed)
+ return 0; // Duplicate section
+
+ //
+ // Pattern matched and tag is expected. Mark the beginning of the section.
+ //
+ location->parsed = 1;
+ location->buffer = test_buffer;
+ location->offset = test_cursor - nx_buffer_base(test_buffer);
+ location->length = 0;
+
+ //
+ // Advance the pointers to consume the whole section.
+ //
+ int consume = 0;
+ unsigned char tag = next_octet(&test_cursor, &test_buffer);
+ if (!test_cursor) return 0;
+ switch (tag) {
+ case 0x45 : // list0
+ break;
+
+ case 0xd0 : // list32
+ case 0xd1 : // map32
+ case 0xb0 : // vbin32
+ consume |= ((int) next_octet(&test_cursor, &test_buffer)) << 24;
+ if (!test_cursor) return 0;
+ consume |= ((int) next_octet(&test_cursor, &test_buffer)) << 16;
+ if (!test_cursor) return 0;
+ consume |= ((int) next_octet(&test_cursor, &test_buffer)) << 8;
+ if (!test_cursor) return 0;
+ // Fall through to the next case...
+
+ case 0xc0 : // list8
+ case 0xc1 : // map8
+ case 0xa0 : // vbin8
+ consume |= (int) next_octet(&test_cursor, &test_buffer);
+ if (!test_cursor) return 0;
+ break;
+ }
+
+ if (consume)
+ advance(&test_cursor, &test_buffer, consume);
+
+ *cursor = test_cursor;
+ *buffer = test_buffer;
+ return 1;
+}
+
+
+static void nx_insert(nx_message_t *msg, const uint8_t *seq, size_t len)
+{
+ nx_buffer_t *buf = DEQ_TAIL(msg->buffers);
+
+ while (len > 0) {
+ if (buf == 0 || nx_buffer_capacity(buf) == 0) {
+ buf = nx_allocate_buffer();
+ if (buf == 0)
+ return;
+ DEQ_INSERT_TAIL(msg->buffers, buf);
+ }
+
+ size_t to_copy = nx_buffer_capacity(buf);
+ if (to_copy > len)
+ to_copy = len;
+ memcpy(nx_buffer_cursor(buf), seq, to_copy);
+ nx_buffer_insert(buf, to_copy);
+ len -= to_copy;
+ seq += to_copy;
+ msg->length += to_copy;
+ }
+}
+
+
+static void nx_insert_8(nx_message_t *msg, uint8_t value)
+{
+ nx_insert(msg, &value, 1);
+}
+
+
+static void nx_insert_32(nx_message_t *msg, uint32_t value)
+{
+ uint8_t buf[4];
+ buf[0] = (uint8_t) ((value & 0xFF000000) >> 24);
+ buf[1] = (uint8_t) ((value & 0x00FF0000) >> 16);
+ buf[2] = (uint8_t) ((value & 0x0000FF00) >> 8);
+ buf[3] = (uint8_t) (value & 0x000000FF);
+ nx_insert(msg, buf, 4);
+}
+
+
+static void nx_insert_64(nx_message_t *msg, uint64_t value)
+{
+ uint8_t buf[8];
+ buf[0] = (uint8_t) ((value & 0xFF00000000000000L) >> 56);
+ buf[1] = (uint8_t) ((value & 0x00FF000000000000L) >> 48);
+ buf[2] = (uint8_t) ((value & 0x0000FF0000000000L) >> 40);
+ buf[3] = (uint8_t) ((value & 0x000000FF00000000L) >> 32);
+ buf[4] = (uint8_t) ((value & 0x00000000FF000000L) >> 24);
+ buf[5] = (uint8_t) ((value & 0x0000000000FF0000L) >> 16);
+ buf[6] = (uint8_t) ((value & 0x000000000000FF00L) >> 8);
+ buf[7] = (uint8_t) (value & 0x00000000000000FFL);
+ nx_insert(msg, buf, 8);
+}
+
+
+static void nx_overwrite(nx_buffer_t **buf, size_t *cursor, uint8_t value)
+{
+ while (*buf) {
+ if (*cursor >= nx_buffer_size(*buf)) {
+ *buf = (*buf)->next;
+ *cursor = 0;
+ } else {
+ nx_buffer_base(*buf)[*cursor] = value;
+ (*cursor)++;
+ return;
+ }
+ }
+}
+
+
+static void nx_overwrite_32(nx_field_location_t *field, uint32_t value)
+{
+ nx_buffer_t *buf = field->buffer;
+ size_t cursor = field->offset;
+
+ nx_overwrite(&buf, &cursor, (uint8_t) ((value & 0xFF000000) >> 24));
+ nx_overwrite(&buf, &cursor, (uint8_t) ((value & 0x00FF0000) >> 24));
+ nx_overwrite(&buf, &cursor, (uint8_t) ((value & 0x0000FF00) >> 24));
+ nx_overwrite(&buf, &cursor, (uint8_t) (value & 0x000000FF));
+}
+
+
+static void nx_start_list_performative(nx_message_t *msg, uint8_t code)
+{
+ //
+ // Insert the short-form performative tag
+ //
+ nx_insert(msg, (const uint8_t*) "\x00\x53", 2);
+ nx_insert_8(msg, code);
+
+ //
+ // Open the list with a list32 tag
+ //
+ nx_insert_8(msg, 0xd0);
+
+ //
+ // Mark the current location to later overwrite the length
+ //
+ msg->compose_length.buffer = DEQ_TAIL(msg->buffers);
+ msg->compose_length.offset = nx_buffer_size(msg->compose_length.buffer);
+ msg->compose_length.length = 4;
+ msg->compose_length.parsed = 1;
+
+ nx_insert(msg, (const uint8_t*) "\x00\x00\x00\x00", 4);
+
+ //
+ // Mark the current location to later overwrite the count
+ //
+ msg->compose_count.buffer = DEQ_TAIL(msg->buffers);
+ msg->compose_count.offset = nx_buffer_size(msg->compose_count.buffer);
+ msg->compose_count.length = 4;
+ msg->compose_count.parsed = 1;
+
+ nx_insert(msg, (const uint8_t*) "\x00\x00\x00\x00", 4);
+
+ msg->length = 4; // Include the length of the count field
+ msg->count = 0;
+}
+
+
+static void nx_end_list(nx_message_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)
+{
+ 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;
+}
+
+
+void nx_allocator_initialize(const nx_allocator_config_t *c)
+{
+ config = c;
+
+ // Initialize the fields in the global structure.
+ DEQ_INIT(global.message_free_list);
+ DEQ_INIT(global.buffer_free_list);
+ global.lock = sys_mutex();
+
+ // Pre-allocate buffers according to the configuration
+ int i;
+ nx_buffer_t *buf;
+
+ 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);
+ }
+}
+
+
+void nx_allocator_finalize(void)
+{
+ // TODO - Free buffers and messages
+}
+
+
+nx_message_t *nx_allocate_message(void)
+{
+ nx_allocator_t *alloc = nx_get_allocator();
+ nx_message_t *msg;
+ int i;
+
+ 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);
+ }
+ }
+ sys_mutex_unlock(global.lock);
+ }
+
+ 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]);
+ }
+ }
+
+ //
+ // If the local free list is still empty, we're out of memory.
+ //
+ if (DEQ_SIZE(alloc->message_free_list) == 0)
+ return 0;
+
+ msg = DEQ_HEAD(alloc->message_free_list);
+ DEQ_REMOVE_HEAD(alloc->message_free_list);
+
+ 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;
+}
+
+
+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);
+ }
+ }
+ 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);
+ }
+
+ if (DEQ_SIZE(alloc->buffer_free_list) == 0)
+ return 0;
+
+ buf = DEQ_HEAD(alloc->buffer_free_list);
+ DEQ_REMOVE_HEAD(alloc->buffer_free_list);
+
+ buf->size = 0;
+
+ return buf;
+}
+
+
+void nx_free_message(nx_message_t *msg)
+{
+ nx_allocator_t *alloc = nx_get_allocator();
+
+ // 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);
+ }
+}
+
+
+void nx_free_buffer(nx_buffer_t *buf)
+{
+ nx_allocator_t *alloc = nx_get_allocator();
+ int i;
+
+ 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);
+ }
+}
+
+
+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;
+
+ //
+ // If there is no message associated with the delivery, this is the first time
+ // we've received anything on this delivery. Allocate a message descriptor and
+ // link it and the delivery together.
+ //
+ if (!msg) {
+ msg = 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.
+ //
+ if (!pn_delivery_settled(delivery))
+ msg->in_delivery = delivery;
+ }
+
+ //
+ // Get a reference to the tail buffer on the message. This is the buffer into which
+ // 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);
+ if (!buf) {
+ buf = nx_allocate_buffer();
+ DEQ_INSERT_TAIL(msg->buffers, buf);
+ }
+
+ while (1) {
+ //
+ // Try to receive enough data to fill the remaining space in the tail buffer.
+ //
+ rc = pn_link_recv(link, (char*) nx_buffer_cursor(buf), nx_buffer_capacity(buf));
+
+ //
+ // If we receive PN_EOS, we have come to the end of the message.
+ //
+ if (rc == PN_EOS) {
+ //
+ // If the last buffer in the list is empty, remove it and free it. This
+ // will only happen if the size of the message content is an exact multiple
+ // of the buffer size.
+ //
+ if (nx_buffer_size(buf) == 0) {
+ DEQ_REMOVE_TAIL(msg->buffers);
+ nx_free_buffer(buf);
+ }
+ return msg;
+ }
+
+ if (rc > 0) {
+ //
+ // We have received a positive number of bytes for the message. Advance
+ // the cursor in the buffer.
+ //
+ nx_buffer_insert(buf, rc);
+
+ //
+ // If the buffer is full, allocate a new empty buffer and append it to the
+ // tail of the message's list.
+ //
+ if (nx_buffer_capacity(buf) == 0) {
+ buf = nx_allocate_buffer();
+ DEQ_INSERT_TAIL(msg->buffers, buf);
+ }
+ } else
+ //
+ // We received zero bytes, and no PN_EOS. This means that we've received
+ // all of the data available up to this point, but it does not constitute
+ // the entire message. We'll be back later to finish it up.
+ //
+ break;
+ }
+
+ return NULL;
+}
+
+
+int nx_message_check(nx_message_t *msg, nx_message_depth_t depth)
+{
+
+#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 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 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"
+
+ nx_buffer_t *buffer = DEQ_HEAD(msg->buffers);
+ unsigned char *cursor;
+
+ if (!buffer)
+ return 0; // Invalid - No data in the message
+
+ if (depth == NX_DEPTH_NONE)
+ return 1;
+
+ cursor = nx_buffer_base(buffer);
+
+ //
+ // MESSAGE HEADER
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_LONG, LONG, TAGS_LIST, &msg->section_message_header))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, MSG_HDR_SHORT, SHORT, TAGS_LIST, &msg->section_message_header))
+ return 0;
+
+ if (depth == NX_DEPTH_HEADER)
+ return 1;
+
+ //
+ // DELIVERY ANNOTATION
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_LONG, LONG, TAGS_MAP, &msg->section_delivery_annotation))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, DELIVERY_ANNOTATION_SHORT, SHORT, TAGS_MAP, &msg->section_delivery_annotation))
+ return 0;
+
+ if (depth == NX_DEPTH_DELIVERY_ANNOTATIONS)
+ return 1;
+
+ //
+ // MESSAGE ANNOTATION
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_LONG, LONG, TAGS_MAP, &msg->section_message_annotation))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_ANNOTATION_SHORT, SHORT, TAGS_MAP, &msg->section_message_annotation))
+ return 0;
+
+ if (depth == NX_DEPTH_MESSAGE_ANNOTATIONS)
+ return 1;
+
+ //
+ // MESSAGE PROPERTIES
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_PROPERTIES_LONG, LONG, TAGS_LIST, &msg->section_message_properties))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, MESSAGE_PROPERTIES_SHORT, SHORT, TAGS_LIST, &msg->section_message_properties))
+ return 0;
+
+ if (depth == NX_DEPTH_MESSAGE_PROPERTIES)
+ return 1;
+
+ //
+ // APPLICATION PROPERTIES
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_LONG, LONG, TAGS_MAP, &msg->section_application_properties))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, APPLICATION_PROPERTIES_SHORT, SHORT, TAGS_MAP, &msg->section_application_properties))
+ return 0;
+
+ if (depth == NX_DEPTH_APPLICATION_PROPERTIES)
+ return 1;
+
+ //
+ // 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))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_DATA_SHORT, SHORT, TAGS_BINARY, &msg->section_body))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_LONG, LONG, TAGS_LIST, &msg->section_body))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, BODY_SEQUENCE_SHORT, SHORT, TAGS_LIST, &msg->section_body))
+ return 0;
+
+ if (depth == NX_DEPTH_BODY)
+ return 1;
+
+ //
+ // FOOTER
+ //
+ if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_LONG, LONG, TAGS_MAP, &msg->section_footer))
+ return 0;
+ if (0 == nx_check_and_advance(&buffer, &cursor, FOOTER_SHORT, SHORT, TAGS_MAP, &msg->section_footer))
+ return 0;
+
+ return 1;
+}
+
+
+nx_field_iterator_t *nx_message_field_to(nx_message_t *msg)
+{
+ 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;
+ }
+
+ return 0;
+}
+
+
+nx_field_iterator_t *nx_message_body(nx_message_t *msg)
+{
+ while (1) {
+ if (msg->body.parsed)
+ return nx_field_iterator_buffer(msg->body.buffer, msg->body.offset, msg->body.length, ITER_VIEW_ALL);
+
+ if (msg->section_body.parsed == 0)
+ break;
+
+ nx_buffer_t *buffer = msg->section_body.buffer;
+ unsigned char *cursor = nx_buffer_base(buffer) + msg->section_body.offset;
+ int result;
+
+ result = traverse_field(&cursor, &buffer, &msg->body);
+ if (!result) return 0;
+ }
+
+ return 0;
+}
+
+
+void nx_message_compose_1(nx_message_t *msg, const char *to, nx_buffer_t *buf_chain)
+{
+ nx_message_begin_header(msg);
+ nx_message_insert_boolean(msg, 0); // durable
+ //nx_message_insert_null(msg); // priority
+ //nx_message_insert_null(msg); // ttl
+ //nx_message_insert_boolean(msg, 0); // first-acquirer
+ //nx_message_insert_uint(msg, 0); // delivery-count
+ nx_message_end_header(msg);
+
+ nx_message_begin_message_properties(msg);
+ nx_message_insert_null(msg); // message-id
+ nx_message_insert_null(msg); // user-id
+ nx_message_insert_string(msg, to); // to
+ //nx_message_insert_null(msg); // subject
+ //nx_message_insert_null(msg); // reply-to
+ //nx_message_insert_null(msg); // correlation-id
+ //nx_message_insert_null(msg); // content-type
+ //nx_message_insert_null(msg); // content-encoding
+ //nx_message_insert_timestamp(msg, 0); // absolute-expiry-time
+ //nx_message_insert_timestamp(msg, 0); // creation-time
+ //nx_message_insert_null(msg); // group-id
+ //nx_message_insert_uint(msg, 0); // group-sequence
+ //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);
+}
+
+
+void nx_message_begin_header(nx_message_t *msg)
+{
+ nx_start_list_performative(msg, 0x70);
+}
+
+
+void nx_message_end_header(nx_message_t *msg)
+{
+ nx_end_list(msg);
+}
+
+
+void nx_message_begin_delivery_annotations(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_end_delivery_annotations(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_begin_message_annotations(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_end_message_annotations(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_begin_message_properties(nx_message_t *msg)
+{
+ nx_start_list_performative(msg, 0x73);
+}
+
+
+void nx_message_end_message_properties(nx_message_t *msg)
+{
+ nx_end_list(msg);
+}
+
+
+void nx_message_begin_application_properties(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_end_application_properties(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_append_body_data(nx_message_t *msg, nx_buffer_t *buf_chain)
+{
+ uint32_t len = 0;
+ nx_buffer_t *buf = buf_chain;
+ nx_buffer_t *last = 0;
+ size_t count = 0;
+
+ while (buf) {
+ len += nx_buffer_size(buf);
+ count++;
+ last = buf;
+ buf = DEQ_NEXT(buf);
+ }
+
+ nx_insert(msg, (const uint8_t*) "\x00\x53\x75", 3);
+ if (len < 256) {
+ nx_insert_8(msg, 0xa0); // vbin8
+ nx_insert_8(msg, (uint8_t) len);
+ } else {
+ nx_insert_8(msg, 0xb0); // vbin32
+ nx_insert_32(msg, len);
+ }
+
+ if (len > 0) {
+ buf_chain->prev = msg->buffers.tail;
+ msg->buffers.tail->next = buf_chain;
+ msg->buffers.tail = last;
+ msg->buffers.size += count;
+ }
+}
+
+
+void nx_message_begin_body_sequence(nx_message_t *msg)
+{
+}
+
+
+void nx_message_end_body_sequence(nx_message_t *msg)
+{
+}
+
+
+void nx_message_begin_footer(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_end_footer(nx_message_t *msg)
+{
+ assert(0); // Not Implemented
+}
+
+
+void nx_message_insert_null(nx_message_t *msg)
+{
+ nx_insert_8(msg, 0x40);
+ msg->count++;
+}
+
+
+void nx_message_insert_boolean(nx_message_t *msg, int value)
+{
+ if (value)
+ nx_insert(msg, (const uint8_t*) "\x56\x01", 2);
+ else
+ nx_insert(msg, (const uint8_t*) "\x56\x00", 2);
+ msg->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++;
+}
+
+
+void nx_message_insert_uint(nx_message_t *msg, uint32_t value)
+{
+ if (value == 0) {
+ nx_insert_8(msg, 0x43); // uint0
+ } else if (value < 256) {
+ nx_insert_8(msg, 0x52); // smalluint
+ nx_insert_8(msg, (uint8_t) value);
+ } else {
+ nx_insert_8(msg, 0x70); // uint
+ nx_insert_32(msg, value);
+ }
+ msg->count++;
+}
+
+
+void nx_message_insert_ulong(nx_message_t *msg, uint64_t value)
+{
+ if (value == 0) {
+ nx_insert_8(msg, 0x44); // ulong0
+ } else if (value < 256) {
+ nx_insert_8(msg, 0x53); // smallulong
+ nx_insert_8(msg, (uint8_t) value);
+ } else {
+ nx_insert_8(msg, 0x80); // ulong
+ nx_insert_64(msg, value);
+ }
+ msg->count++;
+}
+
+
+void nx_message_insert_binary(nx_message_t *msg, const uint8_t *start, size_t len)
+{
+ if (len < 256) {
+ nx_insert_8(msg, 0xa0); // vbin8
+ nx_insert_8(msg, (uint8_t) len);
+ } else {
+ nx_insert_8(msg, 0xb0); // vbin32
+ nx_insert_32(msg, len);
+ }
+ nx_insert(msg, start, len);
+ msg->count++;
+}
+
+
+void nx_message_insert_string(nx_message_t *msg, const char *start)
+{
+ 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);
+ } else {
+ nx_insert_8(msg, 0xb1); // str32-utf8
+ nx_insert_32(msg, len);
+ nx_insert(msg, (const uint8_t*) start, len);
+ }
+ msg->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++;
+}
+
+
+void nx_message_insert_symbol(nx_message_t *msg, const char *start, size_t len)
+{
+ if (len < 256) {
+ nx_insert_8(msg, 0xa3); // sym8
+ nx_insert_8(msg, (uint8_t) len);
+ nx_insert(msg, (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);
+ }
+ msg->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++;
+}
+
+
+unsigned char *nx_buffer_base(nx_buffer_t *buf)
+{
+ return (unsigned char*) &buf[1];
+}
+
+
+unsigned char *nx_buffer_cursor(nx_buffer_t *buf)
+{
+ return ((unsigned char*) &buf[1]) + buf->size;
+}
+
+
+size_t nx_buffer_capacity(nx_buffer_t *buf)
+{
+ return config->buffer_size - buf->size;
+}
+
+
+size_t nx_buffer_size(nx_buffer_t *buf)
+{
+ return buf->size;
+}
+
+
+void nx_buffer_insert(nx_buffer_t *buf, size_t len)
+{
+ buf->size += len;
+ assert(buf->size <= config->buffer_size);
+}
+
Propchange: qpid/trunk/qpid/extras/nexus/src/message.c
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/extras/nexus/src/posix/threading.c
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/extras/nexus/src/posix/threading.c?rev=1433735&view=auto
==============================================================================
--- qpid/trunk/qpid/extras/nexus/src/posix/threading.c (added)
+++ qpid/trunk/qpid/extras/nexus/src/posix/threading.c Tue Jan 15 23:28:40 2013
@@ -0,0 +1,126 @@
+/*
+ * 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/nexus/threading.h>
+#include <qpid/nexus/ctools.h>
+#include <stdio.h>
+#include <pthread.h>
+
+struct sys_mutex_t {
+ pthread_mutex_t mutex;
+ int acquired;
+};
+
+sys_mutex_t *sys_mutex(void)
+{
+ sys_mutex_t *mutex = NEW(sys_mutex_t);
+ pthread_mutex_init(&(mutex->mutex), 0);
+ mutex->acquired = 0;
+ return mutex;
+}
+
+
+void sys_mutex_free(sys_mutex_t *mutex)
+{
+ assert(!mutex->acquired);
+ pthread_mutex_destroy(&(mutex->mutex));
+ free(mutex);
+}
+
+
+void sys_mutex_lock(sys_mutex_t *mutex)
+{
+ pthread_mutex_lock(&(mutex->mutex));
+ assert(!mutex->acquired);
+ mutex->acquired++;
+}
+
+
+void sys_mutex_unlock(sys_mutex_t *mutex)
+{
+ mutex->acquired--;
+ assert(!mutex->acquired);
+ pthread_mutex_unlock(&(mutex->mutex));
+}
+
+
+struct sys_cond_t {
+ pthread_cond_t cond;
+};
+
+
+sys_cond_t *sys_cond(void)
+{
+ sys_cond_t *cond = NEW(sys_cond_t);
+ pthread_cond_init(&(cond->cond), 0);
+ return cond;
+}
+
+
+void sys_cond_free(sys_cond_t *cond)
+{
+ pthread_cond_destroy(&(cond->cond));
+ free(cond);
+}
+
+
+void sys_cond_wait(sys_cond_t *cond, sys_mutex_t *held_mutex)
+{
+ assert(held_mutex->acquired);
+ held_mutex->acquired--;
+ pthread_cond_wait(&(cond->cond), &(held_mutex->mutex));
+ held_mutex->acquired++;
+}
+
+
+void sys_cond_signal(sys_cond_t *cond)
+{
+ pthread_cond_signal(&(cond->cond));
+}
+
+
+void sys_cond_signal_all(sys_cond_t *cond)
+{
+ pthread_cond_broadcast(&(cond->cond));
+}
+
+
+struct sys_thread_t {
+ pthread_t thread;
+};
+
+sys_thread_t *sys_thread(void *(*run_function) (void *), void *arg)
+{
+ sys_thread_t *thread = NEW(sys_thread_t);
+ pthread_create(&(thread->thread), 0, run_function, arg);
+ return thread;
+}
+
+
+void sys_thread_free(sys_thread_t *thread)
+{
+ free(thread);
+}
+
+
+void sys_thread_join(sys_thread_t *thread)
+{
+ pthread_join(thread->thread, 0);
+}
+
Propchange: qpid/trunk/qpid/extras/nexus/src/posix/threading.c
------------------------------------------------------------------------------
svn:eol-style = native
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org