Update from trunk r1375509 through r1450773asyncstore

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

1 files changed, 424 insertions, 0 deletions

diff --git a/extras/dispatch/src/router_node.c b/extras/dispatch/src/router_node.c
new file mode 100644
index 0000000000..6ddc8f45dd
--- /dev/null
+++ b/extras/dispatch/src/router_node.c
@@ -0,0 +1,424 @@
+/*
+ * 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 <qpid/dispatch/server.h>
+#include <qpid/dispatch/message.h>
+#include <qpid/dispatch/threading.h>
+#include <qpid/dispatch/timer.h>
+#include <qpid/dispatch/ctools.h>
+#include <qpid/dispatch/hash.h>
+#include <qpid/dispatch/iterator.h>
+#include <qpid/dispatch/log.h>
+#include <qpid/dispatch/router.h>
+
+static char *module="ROUTER_NODE";
+
+struct dx_router_t {
+    dx_node_t          *node;
+    dx_link_list_t      in_links;
+    dx_link_list_t      out_links;
+    dx_message_list_t   in_fifo;
+    sys_mutex_t        *lock;
+    dx_timer_t         *timer;
+    hash_t             *out_hash;
+    uint64_t            dtag;
+};
+
+
+typedef struct {
+    dx_link_t         *link;
+    dx_message_list_t  out_fifo;
+} dx_router_link_t;
+
+
+ALLOC_DECLARE(dx_router_link_t);
+ALLOC_DEFINE(dx_router_link_t);
+
+
+/**
+ * Outbound Delivery Handler
+ */
+static void router_tx_handler(void* context, dx_link_t *link, pn_delivery_t *delivery)
+{
+    dx_router_t      *router  = (dx_router_t*) context;
+    pn_link_t        *pn_link = pn_delivery_link(delivery);
+    dx_router_link_t *rlink   = (dx_router_link_t*) dx_link_get_context(link);
+    dx_message_t     *msg;
+    size_t            size;
+
+    sys_mutex_lock(router->lock);
+    msg = DEQ_HEAD(rlink->out_fifo);
+    if (!msg) {
+        // TODO - Recind the delivery
+        sys_mutex_unlock(router->lock);
+        return;
+    }
+
+    DEQ_REMOVE_HEAD(rlink->out_fifo);
+    size = (DEQ_SIZE(rlink->out_fifo));
+    sys_mutex_unlock(router->lock);
+
+    dx_message_send(msg, pn_link);
+
+    //
+    // If there is no incoming delivery, it was pre-settled.  In this case,
+    // we must pre-settle the outgoing delivery as well.
+    //
+    if (dx_message_in_delivery(msg)) {
+        pn_delivery_set_context(delivery, (void*) msg);
+        dx_message_set_out_delivery(msg, delivery);
+    } else {
+        pn_delivery_settle(delivery);
+        dx_free_message(msg);
+    }
+
+    pn_link_advance(pn_link);
+    pn_link_offered(pn_link, size);
+}
+
+
+/**
+ * Inbound Delivery Handler
+ */
+static void router_rx_handler(void* context, dx_link_t *link, pn_delivery_t *delivery)
+{
+    dx_router_t  *router  = (dx_router_t*) context;
+    pn_link_t    *pn_link = pn_delivery_link(delivery);
+    dx_message_t *msg;
+    int           valid_message = 0;
+
+    //
+    // Receive the message into a local representation.  If the returned message
+    // pointer is NULL, we have not yet received a complete message.
+    //
+    sys_mutex_lock(router->lock);
+    msg = dx_message_receive(delivery);
+    sys_mutex_unlock(router->lock);
+
+    if (!msg)
+        return;
+
+    //
+    // Validate the message through the Properties section
+    //
+    valid_message = dx_message_check(msg, DX_DEPTH_PROPERTIES);
+
+    pn_link_advance(pn_link);
+    pn_link_flow(pn_link, 1);
+
+    if (valid_message) {
+        dx_field_iterator_t *iter = dx_message_field_iterator(msg, DX_FIELD_TO);
+        dx_router_link_t    *rlink;
+        if (iter) {
+            dx_field_iterator_reset(iter, ITER_VIEW_NO_HOST);
+            sys_mutex_lock(router->lock);
+            int result = hash_retrieve(router->out_hash, iter, (void*) &rlink);
+            dx_field_iterator_free(iter);
+
+            if (result == 0) {
+                //
+                // To field is valid and contains a known destination.  Enqueue on
+                // the output fifo for the next-hop-to-destination.
+                //
+                pn_link_t* pn_outlink = dx_link_pn(rlink->link);
+                DEQ_INSERT_TAIL(rlink->out_fifo, msg);
+                pn_link_offered(pn_outlink, DEQ_SIZE(rlink->out_fifo));
+                dx_link_activate(rlink->link);
+            } else {
+                //
+                // To field contains an unknown address.  Release the message.
+                //
+                pn_delivery_update(delivery, PN_RELEASED);
+                pn_delivery_settle(delivery);
+            }
+
+            sys_mutex_unlock(router->lock);
+        }
+    } else {
+        //
+        // Message is invalid.  Reject the message.
+        //
+        pn_delivery_update(delivery, PN_REJECTED);
+        pn_delivery_settle(delivery);
+        pn_delivery_set_context(delivery, 0);
+        dx_free_message(msg);
+    }
+}
+
+
+/**
+ * Delivery Disposition Handler
+ */
+static void router_disp_handler(void* context, dx_link_t *link, pn_delivery_t *delivery)
+{
+    pn_link_t *pn_link = pn_delivery_link(delivery);
+
+    if (pn_link_is_sender(pn_link)) {
+        pn_disposition_t  disp     = pn_delivery_remote_state(delivery);
+        dx_message_t     *msg      = pn_delivery_get_context(delivery);
+        pn_delivery_t    *activate = 0;
+
+        if (msg) {
+            assert(delivery == dx_message_out_delivery(msg));
+            if (disp != 0) {
+                activate = dx_message_in_delivery(msg);
+                pn_delivery_update(activate, disp);
+                // TODO - handling of the data accompanying RECEIVED/MODIFIED
+            }
+
+            if (pn_delivery_settled(delivery)) {
+                //
+                // Downstream delivery has been settled.  Propagate the settlement
+                // upstream.
+                //
+                activate = dx_message_in_delivery(msg);
+                pn_delivery_settle(activate);
+                pn_delivery_settle(delivery);
+                dx_free_message(msg);
+            }
+
+            if (activate) {
+                //
+                // Activate the upstream/incoming link so that the settlement will
+                // get pushed out.
+                //
+                dx_link_t *act_link = (dx_link_t*) pn_link_get_context(pn_delivery_link(activate));
+                dx_link_activate(act_link);
+            }
+
+            return;
+        }
+    }
+
+    pn_delivery_settle(delivery);
+}
+
+
+/**
+ * New Incoming Link Handler
+ */
+static int router_incoming_link_handler(void* context, dx_link_t *link)
+{
+    dx_router_t    *router  = (dx_router_t*) context;
+    dx_link_item_t *item    = new_dx_link_item_t();
+    pn_link_t      *pn_link = dx_link_pn(link);
+
+    if (item) {
+        DEQ_ITEM_INIT(item);
+        item->link = link;
+
+        sys_mutex_lock(router->lock);
+        DEQ_INSERT_TAIL(router->in_links, item);
+        sys_mutex_unlock(router->lock);
+
+        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_open(pn_link);
+    } else {
+        pn_link_close(pn_link);
+    }
+    return 0;
+}
+
+
+/**
+ * New Outgoing Link Handler
+ */
+static int router_outgoing_link_handler(void* context, dx_link_t *link)
+{
+    dx_router_t *router  = (dx_router_t*) context;
+    pn_link_t   *pn_link = dx_link_pn(link);
+    const char  *r_tgt   = pn_terminus_get_address(pn_link_remote_target(pn_link));
+
+    sys_mutex_lock(router->lock);
+    dx_router_link_t *rlink = new_dx_router_link_t();
+    rlink->link = link;
+    DEQ_INIT(rlink->out_fifo);
+    dx_link_set_context(link, rlink);
+
+    dx_field_iterator_t *iter = dx_field_iterator_string(r_tgt, ITER_VIEW_NO_HOST);
+    int result = hash_insert(router->out_hash, iter, rlink);
+    dx_field_iterator_free(iter);
+
+    if (result == 0) {
+        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_open(pn_link);
+        sys_mutex_unlock(router->lock);
+        dx_log(module, LOG_TRACE, "Registered new local address: %s", r_tgt);
+        return 0;
+    }
+
+    dx_log(module, LOG_TRACE, "Address '%s' not registered as it already exists", r_tgt);
+    pn_link_close(pn_link);
+    sys_mutex_unlock(router->lock);
+    return 0;
+}
+
+
+/**
+ * Outgoing Link Writable Handler
+ */
+static int router_writable_link_handler(void* context, dx_link_t *link)
+{
+    dx_router_t      *router = (dx_router_t*) context;
+    int               grant_delivery = 0;
+    pn_delivery_t    *delivery;
+    dx_router_link_t *rlink = (dx_router_link_t*) dx_link_get_context(link);
+    pn_link_t        *pn_link = dx_link_pn(link);
+    uint64_t          tag;
+
+    sys_mutex_lock(router->lock);
+    if (DEQ_SIZE(rlink->out_fifo) > 0) {
+        grant_delivery = 1;
+        tag = router->dtag++;
+    }
+    sys_mutex_unlock(router->lock);
+
+    if (grant_delivery) {
+        pn_delivery(pn_link, pn_dtag((char*) &tag, 8));
+        delivery = pn_link_current(pn_link);
+        if (delivery) {
+            router_tx_handler(context, link, delivery);
+            return 1;
+        }
+    }
+
+    return 0;
+}
+
+
+/**
+ * Link Detached Handler
+ */
+static int router_link_detach_handler(void* context, dx_link_t *link, int closed)
+{
+    dx_router_t    *router  = (dx_router_t*) context;
+    pn_link_t      *pn_link = dx_link_pn(link);
+    const char     *r_tgt   = pn_terminus_get_address(pn_link_remote_target(pn_link));
+    dx_link_item_t *item;
+
+    sys_mutex_lock(router->lock);
+    if (pn_link_is_sender(pn_link)) {
+        item = DEQ_HEAD(router->out_links);
+
+        dx_field_iterator_t *iter = dx_field_iterator_string(r_tgt, ITER_VIEW_NO_HOST);
+        dx_router_link_t    *rlink;
+        if (iter) {
+            int result = hash_retrieve(router->out_hash, iter, (void*) &rlink);
+            if (result == 0) {
+                dx_field_iterator_reset(iter, ITER_VIEW_NO_HOST);
+                hash_remove(router->out_hash, iter);
+                free_dx_router_link_t(rlink);
+                dx_log(module, LOG_TRACE, "Removed local address: %s", r_tgt);
+            }
+            dx_field_iterator_free(iter);
+        }
+    }
+    else
+        item = DEQ_HEAD(router->in_links);
+
+    while (item) {
+        if (item->link == link) {
+            if (pn_link_is_sender(pn_link))
+                DEQ_REMOVE(router->out_links, item);
+            else
+                DEQ_REMOVE(router->in_links, item);
+            free_dx_link_item_t(item);
+            break;
+        }
+        item = item->next;
+    }
+
+    sys_mutex_unlock(router->lock);
+    return 0;
+}
+
+
+static void router_inbound_open_handler(void *type_context, dx_connection_t *conn)
+{
+}
+
+
+static void router_outbound_open_handler(void *type_context, dx_connection_t *conn)
+{
+}
+
+
+static void dx_router_timer_handler(void *context)
+{
+    dx_router_t *router = (dx_router_t*) context;
+
+    //
+    // Periodic processing.
+    //
+    dx_timer_schedule(router->timer, 1000);
+}
+
+
+static dx_node_type_t router_node = {"router", 0, 0,
+                                     router_rx_handler,
+                                     router_tx_handler,
+                                     router_disp_handler,
+                                     router_incoming_link_handler,
+                                     router_outgoing_link_handler,
+                                     router_writable_link_handler,
+                                     router_link_detach_handler,
+                                     0,   // node_created_handler
+                                     0,   // node_destroyed_handler
+                                     router_inbound_open_handler,
+                                     router_outbound_open_handler };
+static int type_registered = 0;
+
+
+dx_router_t *dx_router(dx_router_configuration_t *config)
+{
+    if (!type_registered) {
+        type_registered = 1;
+        dx_container_register_node_type(&router_node);
+    }
+
+    dx_router_t *router = NEW(dx_router_t);
+    dx_container_set_default_node_type(&router_node, (void*) router, DX_DIST_BOTH);
+
+    DEQ_INIT(router->in_links);
+    DEQ_INIT(router->out_links);
+    DEQ_INIT(router->in_fifo);
+
+    router->lock = sys_mutex();
+
+    router->timer = dx_timer(dx_router_timer_handler, (void*) router);
+    dx_timer_schedule(router->timer, 0); // Immediate
+
+    router->out_hash = hash(10, 32, 0);
+    router->dtag = 1;
+
+    return router;
+}
+
+
+void dx_router_free(dx_router_t *router)
+{
+    dx_container_set_default_node_type(0, 0, DX_DIST_BOTH);
+    sys_mutex_free(router->lock);
+    free(router);
+}
+