QPID-5216

  - Fixed propagation of the deletion of locally-attached mobile addresses
  - Changed 'global' address class to 'mobile' to be consistent


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

4 files changed, 94 insertions, 17 deletions

diff --git a/qpid/extras/dispatch/src/hash.c b/qpid/extras/dispatch/src/hash.c
index 24f0999fdc..0cb32acd05 100644
--- a/qpid/extras/dispatch/src/hash.c
+++ b/qpid/extras/dispatch/src/hash.c
@@ -264,9 +264,19 @@ const unsigned char *dx_hash_key_by_handle(const dx_hash_handle_t *handle)
 
 dx_error_t dx_hash_remove_by_handle(dx_hash_t *h, dx_hash_handle_t *handle)
 {
+    unsigned char *key   = 0;
+    dx_error_t     error = dx_hash_remove_by_handle2(h, handle, &key);
+    if (key)
+        free(key);
+    return error;
+}
+
+
+dx_error_t dx_hash_remove_by_handle2(dx_hash_t *h, dx_hash_handle_t *handle, unsigned char **key)
+{
     if (!handle)
         return DX_ERROR_NOT_FOUND;
-    free(handle->item->key);
+    *key = handle->item->key;
     DEQ_REMOVE(handle->bucket->items, handle->item);
     free_dx_hash_item_t(handle->item);
     h->size--;
diff --git a/qpid/extras/dispatch/src/router_node.c b/qpid/extras/dispatch/src/router_node.c
index eefd37cb77..28f22317f3 100644
--- a/qpid/extras/dispatch/src/router_node.c
+++ b/qpid/extras/dispatch/src/router_node.c
@@ -107,18 +107,51 @@ void dx_router_del_node_ref_LH(dx_router_ref_list_t *ref_list, dx_router_node_t
  * Depending on its policy, the address may be eligible for being closed out
  * (i.e. Logging its terminal statistics and freeing its resources).
  */
-void dx_router_check_addr_LH(dx_router_t *router, dx_address_t *addr)
+void dx_router_check_addr(dx_router_t *router, dx_address_t *addr, int was_local)
 {
     if (addr == 0)
         return;
 
-    if (addr->handler || DEQ_SIZE(addr->rlinks) > 0 || DEQ_SIZE(addr->rnodes) > 0)
-        return;
+    unsigned char       *key            = 0;
+    const unsigned char *key_const      = 0;
+    int                  to_delete      = 0;
+    int                  no_more_locals = 0;
+
+    sys_mutex_lock(router->lock);
+    if (addr->handler == 0 && DEQ_SIZE(addr->rlinks) == 0 && DEQ_SIZE(addr->rnodes) == 0)
+        to_delete = 1;
+
+    if (was_local && DEQ_SIZE(addr->rlinks) == 0)
+        no_more_locals = 1;
+
+    if (to_delete) {
+        dx_hash_remove_by_handle2(router->addr_hash, addr->hash_handle, &key);
+        DEQ_REMOVE(router->addrs, addr);
+        dx_hash_handle_free(addr->hash_handle);
+        free_dx_address_t(addr);
+    }
 
-    dx_hash_remove_by_handle(router->addr_hash, addr->hash_handle);
-    DEQ_REMOVE(router->addrs, addr);
-    dx_hash_handle_free(addr->hash_handle);
-    free_dx_address_t(addr);
+    if (!to_delete && no_more_locals)
+        key_const = dx_hash_key_by_handle(addr->hash_handle);
+
+    sys_mutex_unlock(router->lock);
+
+    //
+    // If the address is mobile-class and it was just removed from a local link,
+    // tell the router module that it is no longer attached locally.
+    //
+    if (no_more_locals) {
+        if (key && key[0] == 'M')
+            dx_router_mobile_removed(router, (const char*) key);
+        if (key_const && key_const[0] == 'M')
+            dx_router_mobile_removed(router, (const char*) key_const);
+    }
+
+    //
+    // Free the key that was not freed by the hash table.
+    //
+    if (key)
+        free(key);
 }
 
 
@@ -452,6 +485,16 @@ static void router_rx_handler(void* context, dx_link_t *link, dx_delivery_t *del
 
         if (iter) {
             dx_field_iterator_reset_view(iter, ITER_VIEW_ADDRESS_HASH);
+
+            //
+            // Note: This function is going to need to be refactored so we can put an
+            //       asynchronous address lookup here.  In the event there is a translation
+            //       of the address (via namespace), it will have to be done here after
+            //       obtaining the iterator and before doing the hash lookup.
+            //
+            //       Note that this lookup is only done for global/mobile class addresses.
+            //
+
             dx_hash_retrieve(router->addr_hash, iter, (void*) &addr);
             dx_field_iterator_reset_view(iter, ITER_VIEW_NO_HOST);
             int is_local  = dx_field_iterator_prefix(iter, local_prefix);
@@ -836,7 +879,7 @@ static int router_outgoing_link_handler(void* context, dx_link_t *link)
     sys_mutex_unlock(router->lock);
 
     if (propagate)
-        dx_router_global_added(router, iter);
+        dx_router_mobile_added(router, iter);
 
     if (iter)
         dx_field_iterator_free(iter);
@@ -853,6 +896,7 @@ static int router_link_detach_handler(void* context, dx_link_t *link, int closed
     dx_router_t      *router = (dx_router_t*) context;
     dx_router_link_t *rlink  = (dx_router_link_t*) dx_link_get_context(link);
     dx_router_conn_t *shared = (dx_router_conn_t*) dx_link_get_conn_context(link);
+    dx_address_t     *oaddr  = 0;
 
     if (shared) {
         dx_link_set_conn_context(link, 0);
@@ -869,7 +913,7 @@ static int router_link_detach_handler(void* context, dx_link_t *link, int closed
     //
     if (rlink->link_direction == DX_OUTGOING && rlink->owning_addr) {
         dx_router_del_link_ref_LH(&rlink->owning_addr->rlinks, rlink);
-        dx_router_check_addr_LH(router, rlink->owning_addr);
+        oaddr = rlink->owning_addr;
     }
 
     //
@@ -895,6 +939,11 @@ static int router_link_detach_handler(void* context, dx_link_t *link, int closed
     DEQ_REMOVE(router->links, rlink);
     sys_mutex_unlock(router->lock);
 
+    //
+    // Check to see if the owning address should be deleted
+    //
+    dx_router_check_addr(router, oaddr, 1);
+
     // TODO - wrap the free to handle the recursive items
     free_dx_router_link_t(rlink);
 
diff --git a/qpid/extras/dispatch/src/router_private.h b/qpid/extras/dispatch/src/router_private.h
index 9800189548..066188b3f7 100644
--- a/qpid/extras/dispatch/src/router_private.h
+++ b/qpid/extras/dispatch/src/router_private.h
@@ -163,15 +163,15 @@ struct dx_router_t {
 
 
 
-void dx_router_check_addr_LH(dx_router_t *router, dx_address_t *addr);
+void dx_router_check_addr(dx_router_t *router, dx_address_t *addr, int was_local);
 void dx_router_add_link_ref_LH(dx_router_link_ref_list_t *ref_list, dx_router_link_t *link);
 void dx_router_del_link_ref_LH(dx_router_link_ref_list_t *ref_list, dx_router_link_t *link);
 
 void dx_router_add_node_ref_LH(dx_router_ref_list_t *ref_list, dx_router_node_t *rnode);
 void dx_router_del_node_ref_LH(dx_router_ref_list_t *ref_list, dx_router_node_t *rnode);
 
-void dx_router_global_added(dx_router_t *router, dx_field_iterator_t *iter);
-void dx_router_global_removed(dx_router_t *router, const char *addr);
+void dx_router_mobile_added(dx_router_t *router, dx_field_iterator_t *iter);
+void dx_router_mobile_removed(dx_router_t *router, const char *addr);
 
 
 #endif
diff --git a/qpid/extras/dispatch/src/router_pynode.c b/qpid/extras/dispatch/src/router_pynode.c
index 358fa50447..82b08c0785 100644
--- a/qpid/extras/dispatch/src/router_pynode.c
+++ b/qpid/extras/dispatch/src/router_pynode.c
@@ -430,9 +430,10 @@ static PyObject* dx_unmap_destination(PyObject *self, PyObject *args)
     }
         
     dx_router_del_node_ref_LH(&addr->rnodes, rnode);
-    dx_router_check_addr_LH(router, addr);
     sys_mutex_unlock(router->lock);
 
+    dx_router_check_addr(router, addr, 0);
+
     dx_log(module, LOG_DEBUG, "Remote Destination '%s' Unmapped from router %d", addr_string, maskbit);
 
     Py_INCREF(Py_None);
@@ -631,7 +632,7 @@ void dx_pyrouter_tick(dx_router_t *router)
 }
 
 
-void dx_router_global_added(dx_router_t *router, dx_field_iterator_t *iter)
+void dx_router_mobile_added(dx_router_t *router, dx_field_iterator_t *iter)
 {
     PyObject *pArgs;
     PyObject *pValue;
@@ -641,7 +642,7 @@ void dx_router_global_added(dx_router_t *router, dx_field_iterator_t *iter)
         char *address = (char*) dx_field_iterator_copy(iter);
 
         dx_python_lock();
-        pArgs  = PyTuple_New(1);
+        pArgs = PyTuple_New(1);
         PyTuple_SetItem(pArgs, 0, PyString_FromString(address));
         pValue = PyObject_CallObject(router->pyAdded, pArgs);
         if (PyErr_Occurred()) {
@@ -658,7 +659,24 @@ void dx_router_global_added(dx_router_t *router, dx_field_iterator_t *iter)
 }
 
 
-void dx_router_global_removed(dx_router_t *router, const char *addr)
+void dx_router_mobile_removed(dx_router_t *router, const char *address)
 {
+    PyObject *pArgs;
+    PyObject *pValue;
+
+    if (router->pyRemoved && router->router_mode == DX_ROUTER_MODE_INTERIOR) {
+        dx_python_lock();
+        pArgs = PyTuple_New(1);
+        PyTuple_SetItem(pArgs, 0, PyString_FromString(address));
+        pValue = PyObject_CallObject(router->pyRemoved, pArgs);
+        if (PyErr_Occurred()) {
+            PyErr_Print();
+        }
+        Py_DECREF(pArgs);
+        if (pValue) {
+            Py_DECREF(pValue);
+        }
+        dx_python_unlock();
+    }
 }