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 2016/03/19 00:07:07 UTC
[49/50] [abbrv] qpid-dispatch git commit: DISPATCH-179 - Removed dead
files
DISPATCH-179 - Removed dead files
Project: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/commit/2291e5cd
Tree: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/tree/2291e5cd
Diff: http://git-wip-us.apache.org/repos/asf/qpid-dispatch/diff/2291e5cd
Branch: refs/heads/master
Commit: 2291e5cd4ce7128a6e9fcabae80958042fa009a2
Parents: 51d0793
Author: Ted Ross <tr...@redhat.com>
Authored: Fri Mar 18 16:43:53 2016 -0400
Committer: Ted Ross <tr...@redhat.com>
Committed: Fri Mar 18 16:43:53 2016 -0400
----------------------------------------------------------------------
src/dispatch.c | 1 -
src/lrp.c | 248 ----------------------------------
src/lrp_private.h | 40 ------
src/router_config.c | 1 -
src/router_delivery.c | 210 -----------------------------
src/router_forwarders.c | 286 ---------------------------------------
src/router_node.c | 1 -
src/router_pynode.c | 1 -
src/waypoint.c | 309 -------------------------------------------
src/waypoint_private.h | 61 ---------
10 files changed, 1158 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/dispatch.c
----------------------------------------------------------------------
diff --git a/src/dispatch.c b/src/dispatch.c
index d4cdafe..c05d24c 100644
--- a/src/dispatch.c
+++ b/src/dispatch.c
@@ -29,7 +29,6 @@
#include "alloc.h"
#include "log_private.h"
#include "router_private.h"
-#include "waypoint_private.h"
#include "message_private.h"
#include "entity.h"
#include "entity_cache.h"
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/lrp.c
----------------------------------------------------------------------
diff --git a/src/lrp.c b/src/lrp.c
deleted file mode 100644
index d3d0b47..0000000
--- a/src/lrp.c
+++ /dev/null
@@ -1,248 +0,0 @@
-/*
- * 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 "dispatch_private.h"
-#include "entity_cache.h"
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/connection_manager.h>
-#include <qpid/dispatch/timer.h>
-#include <memory.h>
-#include <stdio.h>
-
-static const char qd_link_route_addr_prefix_inbound = 'C';
-static const char qd_link_route_addr_prefix_outbound = 'D';
-
-
-//
-// DEPRECATE
-//
-static void qd_lrpc_open_handler(void *context, qd_connection_t *conn)
-{
- qd_lrp_container_t *lrpc = (qd_lrp_container_t*) context;
- qd_lrp_t *lrp = DEQ_HEAD(lrpc->lrps);
- qd_router_t *router = lrpc->qd->router;
-
- lrpc->conn = conn;
-
- while (lrp) {
- qd_address_t *addr;
- qd_field_iterator_t *iter;
- //bool propagate;
- char unused;
-
- qd_log(router->log_source, QD_LOG_INFO, "Activating Prefix '%s' for routed links to '%s'",
- lrp->prefix, qd_config_connector_name(lrpc->cc));
-
- //
- // Create an address iterator for the prefix address with the namespace
- // prefix for link-attach routed addresses.
- //
- iter = qd_address_iterator_string(lrp->prefix, ITER_VIEW_ADDRESS_HASH);
-
- if (lrp->inbound) {
- qd_address_iterator_override_prefix(iter, qd_link_route_addr_prefix_inbound);
-
- //
- // Find the address in the router's hash table. If not found, create one
- // and hash it into the table.
- //
- sys_mutex_lock(router->lock);
- qd_hash_retrieve(router->addr_hash, iter, (void**) &addr);
- if (!addr) {
- addr = qd_address(router_semantics_for_addr(router, iter, '\0', &unused));
- qd_hash_insert(router->addr_hash, iter, addr, &addr->hash_handle);
- DEQ_INSERT_TAIL(router->addrs, addr);
- qd_entity_cache_add(QD_ROUTER_ADDRESS_TYPE, addr);
- }
-
- //
- // Link the LRP record into the address as a local endpoint for routed link-attaches.
- // If this is the first instance for this address, flag the address for propagation
- // across the network.
- //
- qd_router_add_lrp_ref_LH(&addr->lrps, lrp);
- //propagate = DEQ_SIZE(addr->lrps) == 1;
- sys_mutex_unlock(router->lock);
-
- //
- // Propagate the address if appropriate
- //
- //if (propagate)
- // qd_router_mobile_added(router, iter);
- }
-
- if (lrp->outbound) {
- qd_address_iterator_reset_view(iter, ITER_VIEW_ADDRESS_HASH);
- qd_address_iterator_override_prefix(iter, qd_link_route_addr_prefix_outbound);
-
- //
- // Find the address in the router's hash table. If not found, create one
- // and hash it into the table.
- //
- sys_mutex_lock(router->lock);
- qd_hash_retrieve(router->addr_hash, iter, (void**) &addr);
- if (!addr) {
- addr = qd_address(router_semantics_for_addr(router, iter, '\0', &unused));
- qd_hash_insert(router->addr_hash, iter, addr, &addr->hash_handle);
- DEQ_INSERT_TAIL(router->addrs, addr);
- qd_entity_cache_add(QD_ROUTER_ADDRESS_TYPE, addr);
- }
-
- //
- // Link the LRP record into the address as a local endpoint for routed link-attaches.
- // If this is the first instance for this address, flag the address for propagation
- // across the network.
- //
- qd_router_add_lrp_ref_LH(&addr->lrps, lrp);
- //propagate = DEQ_SIZE(addr->lrps) == 1;
- sys_mutex_unlock(router->lock);
-
- //
- // Propagate the address if appropriate
- //
- //if (propagate)
- // qd_router_mobile_added(router, iter);
- }
-
- qd_field_iterator_free(iter);
- lrp = DEQ_NEXT(lrp);
- }
-}
-
-
-static void qd_lrpc_close_handler(void *context, qd_connection_t *conn)
-{
- qd_lrp_container_t *lrpc = (qd_lrp_container_t*) context;
- qd_lrp_t *lrp = DEQ_HEAD(lrpc->lrps);
- qd_router_t *router = lrpc->qd->router;
-
- lrpc->conn = 0;
-
- while (lrp) {
- qd_address_t *addr;
- qd_field_iterator_t *iter;
- bool propagate = false;
-
- qd_log(router->log_source, QD_LOG_INFO, "Removing Prefix '%s' for routed links to '%s'",
- lrp->prefix, qd_config_connector_name(lrpc->cc));
-
- //
- // Create an address iterator for the prefix address with the namespace
- // prefix for link-attach routed addresses.
- //
- iter = qd_address_iterator_string(lrp->prefix, ITER_VIEW_ADDRESS_HASH);
-
- if (lrp->inbound) {
- //
- // Find the address in the router's hash table.
- //
- qd_address_iterator_override_prefix(iter, qd_link_route_addr_prefix_inbound);
- sys_mutex_lock(router->lock);
- qd_hash_retrieve(router->addr_hash, iter, (void**) &addr);
- if (addr) {
- //
- // Unlink the lrp from the address. If this is the last lrp in the address, we need
- // to tell the other routers.
- //
- qd_router_del_lrp_ref_LH(&addr->lrps, lrp);
- propagate = DEQ_SIZE(addr->lrps) == 0;
- }
- sys_mutex_unlock(router->lock);
-
- //
- // Propagate the address if appropriate
- //
- if (propagate) {
- char *text = (char*) qd_field_iterator_copy(iter);
- //qd_router_mobile_removed(router, text);
- free(text);
- }
- }
-
- if (lrp->outbound) {
- //
- // Find the address in the router's hash table.
- //
- qd_address_iterator_reset_view(iter, ITER_VIEW_ADDRESS_HASH);
- qd_address_iterator_override_prefix(iter, qd_link_route_addr_prefix_outbound);
- sys_mutex_lock(router->lock);
- qd_hash_retrieve(router->addr_hash, iter, (void**) &addr);
- if (addr) {
- //
- // Unlink the lrp from the address. If this is the last lrp in the address, we need
- // to tell the other routers.
- //
- qd_router_del_lrp_ref_LH(&addr->lrps, lrp);
- propagate = DEQ_SIZE(addr->lrps) == 0;
- }
- sys_mutex_unlock(router->lock);
-
- //
- // Propagate the address if appropriate
- //
- if (propagate) {
- char *text = (char*) qd_field_iterator_copy(iter);
- //qd_router_mobile_removed(router, text);
- free(text);
- }
- }
-
- qd_field_iterator_free(iter);
- lrp = DEQ_NEXT(lrp);
- }
-}
-
-
-void qd_lrpc_timer_handler(void *context)
-{
- qd_lrp_container_t *lrpc = (qd_lrp_container_t*) context;
- qd_connection_manager_set_handlers(lrpc->cc,
- qd_lrpc_open_handler,
- qd_lrpc_close_handler,
- (void*) lrpc);
- qd_connection_manager_start_on_demand(lrpc->qd, lrpc->cc);
-}
-
-
-qd_lrp_t *qd_lrp_LH(const char *prefix, bool inbound, bool outbound, qd_lrp_container_t *lrpc)
-{
- qd_lrp_t *lrp = NEW(qd_lrp_t);
-
- if (lrp) {
- DEQ_ITEM_INIT(lrp);
- lrp->prefix = strdup(prefix);
- lrp->inbound = inbound;
- lrp->outbound = outbound;
- lrp->container = lrpc;
- DEQ_INSERT_TAIL(lrpc->lrps, lrp);
- }
-
- return lrp;
-}
-
-
-void qd_lrp_free(qd_lrp_t *lrp)
-{
- if (lrp) {
- free(lrp->prefix);
- DEQ_REMOVE(lrp->container->lrps, lrp);
- free(lrp);
- }
-}
-
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/lrp_private.h
----------------------------------------------------------------------
diff --git a/src/lrp_private.h b/src/lrp_private.h
deleted file mode 100644
index 6dda219..0000000
--- a/src/lrp_private.h
+++ /dev/null
@@ -1,40 +0,0 @@
-#ifndef __lrp_private_h__
-#define __lrp_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 "dispatch_private.h"
-
-/**
- * @file
- *
- * An link-route-pattern is an object that defines an address prefix for link-routing
- * and associates that prefix with an on-demand connector.
- *
- * The address prefix is propagated across the network and can be used as a target for
- * routed links destined for this LRP.
- */
-
-qd_lrp_t *qd_lrp_LH(const char *prefix, bool inbound, bool outbound, qd_lrp_container_t *lrpc);
-
-void qd_lrp_free(qd_lrp_t *lrp);
-
-void qd_lrpc_timer_handler(void *context);
-
-#endif
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/router_config.c
----------------------------------------------------------------------
diff --git a/src/router_config.c b/src/router_config.c
index c6963e8..962b5b4 100644
--- a/src/router_config.c
+++ b/src/router_config.c
@@ -22,7 +22,6 @@
#include <qpid/dispatch/log.h>
#include "dispatch_private.h"
#include "router_private.h"
-#include "lrp_private.h"
#include "entity.h"
#include "entity_cache.h"
#include "schema_enum.h"
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/router_delivery.c
----------------------------------------------------------------------
diff --git a/src/router_delivery.c b/src/router_delivery.c
deleted file mode 100644
index 44f9581..0000000
--- a/src/router_delivery.c
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * 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.
- */
-/** Encapsulates a proton message delivery */
-
-#include <qpid/dispatch.h>
-#include "dispatch_private.h"
-#include "router_private.h"
-
-struct qd_router_delivery_t {
- DEQ_LINKS(qd_router_delivery_t);
- pn_delivery_t *pn_delivery;
- qd_router_delivery_t *peer;
- void *context;
- uint64_t disposition;
- qd_router_link_t *link;
- int in_fifo;
- bool pending_delete;
-};
-ALLOC_DECLARE(qd_router_delivery_t);
-ALLOC_DEFINE(qd_router_delivery_t);
-
-// create a router delivery from a proton delivery received on rlink
-qd_router_delivery_t *qd_router_delivery(qd_router_link_t *rlink, pn_delivery_t *pnd)
-{
- assert(pn_delivery_get_context(pnd) == 0);
- qd_router_delivery_t *delivery = new_qd_router_delivery_t();
- if (delivery) {
- DEQ_ITEM_INIT(delivery);
- delivery->pn_delivery = pnd;
- delivery->peer = 0;
- delivery->context = 0;
- delivery->disposition = 0;
- delivery->link = rlink;
- delivery->in_fifo = 0;
- delivery->pending_delete = false;
- DEQ_INSERT_TAIL(rlink->deliveries, delivery);
- pn_delivery_set_context(pnd, delivery);
- }
-
- return delivery;
-}
-
-
-// generate a new router delivery for rlink
-qd_router_delivery_t *qd_router_link_new_delivery(qd_router_link_t *rlink, pn_delivery_tag_t tag)
-{
- qd_link_t *link = rlink->link;
- pn_link_t *pnl = qd_link_pn(link);
-
- //
- // If there is a current delivery on this outgoing link, something
- // is wrong with the delivey algorithm. We assume that the current
- // delivery ('pnd' below) is the one created by pn_delivery. If it is
- // not, then my understanding of how proton works is incorrect.
- //
- assert(!pn_link_current(pnl));
-
- pn_delivery(pnl, tag);
- pn_delivery_t *pnd = pn_link_current(pnl);
-
- if (!pnd)
- return 0;
-
- return qd_router_delivery(rlink, pnd);
-}
-
-// mark the delivery as 'undeliverable-here' so peers won't re-forward it to
-// us.
-void qd_router_delivery_set_undeliverable_LH(qd_router_delivery_t *delivery)
-{
- if (delivery->pn_delivery) {
- pn_disposition_t *dp = pn_delivery_local(delivery->pn_delivery);
- if (dp) {
- pn_disposition_set_undeliverable(dp, true);
- }
- }
-}
-
-void qd_router_delivery_free_LH(qd_router_delivery_t *delivery, uint64_t final_disposition)
-{
- if (delivery->pn_delivery) {
- if (final_disposition > 0)
- pn_delivery_update(delivery->pn_delivery, final_disposition);
- pn_delivery_set_context(delivery->pn_delivery, 0);
- pn_delivery_settle(delivery->pn_delivery);
- delivery->pn_delivery = 0;
- }
-
- if (delivery->peer)
- qd_router_delivery_unlink_LH(delivery);
-
- if (delivery->link) {
- DEQ_REMOVE(delivery->link->deliveries, delivery);
- delivery->link = 0;
- }
- if (delivery->in_fifo)
- delivery->pending_delete = true;
- else {
- free_qd_router_delivery_t(delivery);
- }
-}
-
-
-void qd_router_delivery_link_peers_LH(qd_router_delivery_t *right, qd_router_delivery_t *left)
-{
- right->peer = left;
- left->peer = right;
-}
-
-
-void qd_router_delivery_unlink_LH(qd_router_delivery_t *delivery)
-{
- if (delivery->peer) {
- delivery->peer->peer = 0;
- delivery->peer = 0;
- }
-}
-
-
-void qd_router_delivery_fifo_enter_LH(qd_router_delivery_t *delivery)
-{
- delivery->in_fifo++;
-}
-
-
-bool qd_router_delivery_fifo_exit_LH(qd_router_delivery_t *delivery)
-{
- delivery->in_fifo--;
- if (delivery->in_fifo == 0 && delivery->pending_delete) {
- free_qd_router_delivery_t(delivery);
- return false;
- }
-
- return true;
-}
-
-
-void qd_router_delivery_set_context(qd_router_delivery_t *delivery, void *context)
-{
- delivery->context = context;
-}
-
-
-void *qd_router_delivery_context(qd_router_delivery_t *delivery)
-{
- return delivery->context;
-}
-
-
-qd_router_delivery_t *qd_router_delivery_peer(qd_router_delivery_t *delivery)
-{
- return delivery->peer;
-}
-
-
-pn_delivery_t *qd_router_delivery_pn(qd_router_delivery_t *delivery)
-{
- return delivery->pn_delivery;
-}
-
-
-void qd_router_delivery_settle(qd_router_delivery_t *delivery)
-{
- if (delivery->pn_delivery) {
- pn_delivery_set_context(delivery->pn_delivery, 0);
- pn_delivery_settle(delivery->pn_delivery);
- delivery->pn_delivery = 0;
- }
-}
-
-
-bool qd_router_delivery_settled(qd_router_delivery_t *delivery)
-{
- return pn_delivery_settled(delivery->pn_delivery);
-}
-
-
-bool qd_router_delivery_disp_changed(qd_router_delivery_t *delivery)
-{
- return delivery->disposition != pn_delivery_remote_state(delivery->pn_delivery);
-}
-
-
-uint64_t qd_router_delivery_disp(qd_router_delivery_t *delivery)
-{
- delivery->disposition = pn_delivery_remote_state(delivery->pn_delivery);
- return delivery->disposition;
-}
-
-
-qd_router_link_t *qd_router_delivery_link(qd_router_delivery_t *delivery)
-{
- return delivery->link;
-}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/router_forwarders.c
----------------------------------------------------------------------
diff --git a/src/router_forwarders.c b/src/router_forwarders.c
deleted file mode 100644
index 05bb9d0..0000000
--- a/src/router_forwarders.c
+++ /dev/null
@@ -1,286 +0,0 @@
-/*
- * 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 "dispatch_private.h"
-
-/** defines a default set of forwarding behaviors based on the treatment of an
- * address.
- */
-
-
-static void forward_to_direct_subscribers_LH(qd_address_t *addr,
- qd_router_delivery_t *delivery,
- qd_message_t *msg,
- int *fanout)
-{
- qd_router_link_ref_t *dest_link_ref = DEQ_HEAD(addr->rlinks);
- while (dest_link_ref) {
- qd_routed_event_t *re = new_qd_routed_event_t();
- DEQ_ITEM_INIT(re);
- re->delivery = 0;
- re->message = qd_message_copy(msg);
- re->settle = 0;
- re->disposition = 0;
- DEQ_INSERT_TAIL(dest_link_ref->link->msg_fifo, re);
-
- (*fanout)++;
- if (*fanout == 1) {
- re->delivery = delivery;
- qd_router_delivery_fifo_enter_LH(delivery);
- }
-
- addr->deliveries_egress++;
- qd_link_activate(dest_link_ref->link->link);
-
- //
- // If the fanout is single, exit the loop here. We only want to send one message copy.
- //
- if (QD_FANOUT(addr->treatment) == QD_FANOUT_SINGLE)
- break;
-
- dest_link_ref = DEQ_NEXT(dest_link_ref);
- }
-
- //
- // If dest_link_ref is not null here, we exited after sending one message copy.
- // If the number of local links is greater than one, rotate the head link to the
- // tail so we balance the message deliveries.
- //
- if (dest_link_ref && DEQ_SIZE(addr->rlinks) > 1) {
- assert(DEQ_HEAD(addr->rlinks) == dest_link_ref);
- DEQ_REMOVE_HEAD(addr->rlinks);
- DEQ_INSERT_TAIL(addr->rlinks, dest_link_ref);
- }
-}
-
-
-static void forward_to_remote_subscribers_LH(qd_router_t *router,
- qd_address_t *addr,
- qd_router_delivery_t *delivery,
- qd_message_t *msg,
- int *fanout,
- qd_field_iterator_t *ingress_iter)
-{
- //
- // Get the mask bit associated with the ingress router for the message.
- // This will be compared against the "valid_origin" masks for each
- // candidate destination router.
- //
- int origin = -1;
- if (ingress_iter && !(addr->treatment & QD_BYPASS_VALID_ORIGINS)) {
- qd_address_iterator_reset_view(ingress_iter, ITER_VIEW_NODE_HASH);
- qd_address_t *origin_addr;
- qd_hash_retrieve(router->addr_hash, ingress_iter, (void*) &origin_addr);
- if (origin_addr && DEQ_SIZE(origin_addr->rnodes) == 1) {
- qd_router_ref_t *rref = DEQ_HEAD(origin_addr->rnodes);
- origin = rref->router->mask_bit;
- }
- } else
- origin = 0;
-
- //
- // Forward to the next-hops for remote destinations.
- //
- if (origin >= 0) {
- qd_router_ref_t *dest_node_ref = DEQ_HEAD(addr->rnodes);
- qd_router_link_t *dest_link;
- qd_bitmask_t *link_set = qd_bitmask(0);
-
- //
- // Loop over the target nodes for this address. Build a set of outgoing links
- // for which there are valid targets. We do this to avoid sending more than one
- // message down a given link. It's possible that there are multiple destinations
- // for this address that are all reachable over the same link. In this case, we
- // will send only one copy of the message over the link and allow a downstream
- // router to fan the message out.
- //
- while (dest_node_ref) {
- if (dest_node_ref->router->next_hop)
- dest_link = dest_node_ref->router->next_hop->peer_link;
- else
- dest_link = dest_node_ref->router->peer_link;
- if (dest_link && qd_bitmask_value(dest_node_ref->router->valid_origins, origin))
- qd_bitmask_set_bit(link_set, dest_link->mask_bit);
- dest_node_ref = DEQ_NEXT(dest_node_ref);
- }
-
- //
- // Send a copy of the message outbound on each identified link.
- //
- int link_bit;
- while (qd_bitmask_first_set(link_set, &link_bit)) {
- qd_bitmask_clear_bit(link_set, link_bit);
- dest_link = router->out_links_by_mask_bit[link_bit];
- if (dest_link) {
- qd_routed_event_t *re = new_qd_routed_event_t();
- DEQ_ITEM_INIT(re);
- re->delivery = 0;
- re->message = qd_message_copy(msg);
- re->settle = 0;
- re->disposition = 0;
- DEQ_INSERT_TAIL(dest_link->msg_fifo, re);
-
- (*fanout)++;
- if (*fanout == 1) {
- re->delivery = delivery;
- qd_router_delivery_fifo_enter_LH(delivery);
- }
-
- addr->deliveries_transit++;
- qd_link_activate(dest_link->link);
-
- //
- // If the fanout is single, exit the loop here. We only want to send one message copy.
- //
- if (QD_FANOUT(addr->semantics) == QD_FANOUT_SINGLE)
- break;
- }
- }
-
- qd_bitmask_free(link_set);
- }
-}
-
-
-/** Multicast forwarder:
- */
-static bool forwarder_multicast_LH(qd_router_forwarder_t *forwarder,
- qd_router_t *router,
- qd_message_t *msg,
- qd_router_delivery_t *delivery,
- qd_address_t *addr,
- qd_field_iterator_t *ingress_iter,
- bool is_direct)
-{
- int fanout = 0;
- //
- // Forward to all of the local links receiving this address.
- //
- forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout);
-
- //
- // If the address form is direct to this router node, don't relay it on to
- // any other part of the network.
- //
- if (!is_direct)
- forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter);
-
- return fanout != 0;
-}
-
-
-/** Forward using the 'closest' bias:
- */
-static bool forwarder_anycast_closest_LH(qd_router_forwarder_t *forwarder,
- qd_router_t *router,
- qd_message_t *msg,
- qd_router_delivery_t *delivery,
- qd_address_t *addr,
- qd_field_iterator_t *ingress_iter,
- bool is_direct)
-{
- int fanout = 0;
- //
- // First, try to find a directly connected consumer for the address. If
- // there is none, then look for the closest remote consumer.
- //
- forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout);
- if (fanout == 0 && !is_direct)
- forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter);
-
- return fanout != 0;
-}
-
-
-/** Forwarding using a 'balanced' bias:
- */
-static bool forwarder_anycast_balanced_LH(qd_router_forwarder_t *forwarder,
- qd_router_t *router,
- qd_message_t *msg,
- qd_router_delivery_t *delivery,
- qd_address_t *addr,
- qd_field_iterator_t *ingress_iter,
- bool is_direct)
-{
- int fanout = 0;
- //
- // Alternate between looking first for a local consumer and looking first
- // for a remote consumer.
- //
- addr->toggle = !addr->toggle;
- if (addr->toggle) {
- forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout);
- if (fanout == 0 && !is_direct)
- forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter);
- } else {
- if (!is_direct)
- forward_to_remote_subscribers_LH(router, addr, delivery, msg, &fanout, ingress_iter);
- if (fanout == 0)
- forward_to_direct_subscribers_LH(addr, delivery, msg, &fanout);
- }
-
- return fanout != 0;
-}
-
-
-/* release method for default forwarders:
- */
-static void forwarder_release(qd_router_forwarder_t *forwarder)
-{
- // no-op - they're static singletons!
-}
-
-
-/* The default forwarders:
- */
-static qd_router_forwarder_t multicast_forwarder = {
- forwarder_multicast_LH, /* forward method */
- forwarder_release,
-};
-static qd_router_forwarder_t anycast_closest_forwarder = {
- forwarder_anycast_closest_LH, /* forward method */
- forwarder_release,
-};
-static qd_router_forwarder_t anycast_balanced_forwarder = {
- forwarder_anycast_balanced_LH, /* forward method */
- forwarder_release,
-};
-
-
-/** Get the proper default forwarder for an address of the given treatment:
- */
-qd_router_forwarder_t *qd_router_get_forwarder(qd_address_treatment_t treatment)
-{
- switch (QD_FANOUT(treatment)) {
- case QD_FANOUT_MULTIPLE:
- return &multicast_forwarder;
- case QD_FANOUT_SINGLE:
- switch (QD_BIAS(treatment)) {
- case QD_BIAS_CLOSEST:
- return &anycast_closest_forwarder;
- case QD_BIAS_SPREAD:
- return &anycast_balanced_forwarder;
- }
- }
- assert(false); // invalid treatment? need new forwarder?
- return 0;
-}
-
-
-
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/router_node.c
----------------------------------------------------------------------
diff --git a/src/router_node.c b/src/router_node.c
index b374cd7..bf093df 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -26,7 +26,6 @@
#include "dispatch_private.h"
#include "entity_cache.h"
#include "router_private.h"
-#include "waypoint_private.h"
const char *QD_ROUTER_NODE_TYPE = "router.node";
const char *QD_ROUTER_ADDRESS_TYPE = "router.address";
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/router_pynode.c
----------------------------------------------------------------------
diff --git a/src/router_pynode.c b/src/router_pynode.c
index 4d2c819..db95d9b 100644
--- a/src/router_pynode.c
+++ b/src/router_pynode.c
@@ -25,7 +25,6 @@
#include <qpid/dispatch.h>
#include "dispatch_private.h"
#include "router_private.h"
-#include "waypoint_private.h"
#include "entity_cache.h"
static qd_log_source_t *log_source = 0;
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/waypoint.c
----------------------------------------------------------------------
diff --git a/src/waypoint.c b/src/waypoint.c
deleted file mode 100644
index bfff845..0000000
--- a/src/waypoint.c
+++ /dev/null
@@ -1,309 +0,0 @@
-/*
- * 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 "waypoint_private.h"
-#include "dispatch_private.h"
-#include "router_private.h"
-#include "entity_cache.h"
-#include <qpid/dispatch/ctools.h>
-#include <qpid/dispatch/threading.h>
-#include <qpid/dispatch/connection_manager.h>
-#include <memory.h>
-#include <stdio.h>
-
-typedef struct qd_waypoint_ref_t qd_waypoint_ref_t;
-typedef struct qd_waypoint_context_t qd_waypoint_context_t;
-
-struct qd_waypoint_ref_t {
- DEQ_LINKS(qd_waypoint_ref_t);
- qd_waypoint_t *wp;
-};
-
-DEQ_DECLARE(qd_waypoint_ref_t, qd_waypoint_ref_list_t);
-
-struct qd_waypoint_context_t {
- qd_waypoint_ref_list_t refs;
-};
-
-// Convenience for logging waypoint messages, expects qd and wp to be defined.
-#define LOG(LEVEL, MSG, ...) qd_log(qd->router->log_source, QD_LOG_##LEVEL, "waypoint=%s: " MSG, wp->address, __VA_ARGS__)
-
-static void qd_waypoint_visit_sink_LH(qd_dispatch_t *qd, qd_waypoint_t *wp)
-{
- qd_router_t *router = qd->router;
- qd_address_t *addr = wp->in_address;
- char unused;
-
- //
- // If the waypoint has no in-address, look it up in the hash table or create
- // a new one and put it in the hash table.
- //
- if (!addr) {
- //
- // Compose the phased-address and search the routing table for the address.
- // If it's not found, add it to the table but leave the link/router linkages empty.
- //
- qd_field_iterator_t *iter = qd_address_iterator_string(wp->address, ITER_VIEW_ADDRESS_HASH);
- qd_address_iterator_set_phase(iter, wp->in_phase);
- qd_hash_retrieve(router->addr_hash, iter, (void*) &addr);
-
- if (!addr) {
- addr = qd_address(router_semantics_for_addr(router, iter, wp->in_phase, &unused));
- qd_hash_insert(router->addr_hash, iter, addr, &addr->hash_handle);
- DEQ_INSERT_TAIL(router->addrs, addr);
- addr->waypoint = true;
- qd_entity_cache_add(QD_ROUTER_ADDRESS_TYPE, addr);
- }
-
- wp->in_address = addr;
- qd_field_iterator_free(iter);
- LOG(TRACE, "Sink in-address=%s, in-phase=%c",
- qd_address_logstr(wp->in_address), wp->in_phase);
- }
-
- if (!wp->connected) {
- LOG(TRACE, "Sink start connector %s", wp->connector_name);
- qd_connection_manager_start_on_demand(qd, wp->connector);
- }
- else if (!wp->out_link) {
- wp->out_link = qd_link(router->node, wp->connection, QD_OUTGOING, wp->address);
- pn_terminus_set_address(qd_link_target(wp->out_link), wp->address);
-
- qd_router_link_t *rlink = 0; //qd_router_link(wp->out_link, QD_LINK_WAYPOINT, QD_OUTGOING, addr, wp, 0);
-
- qd_entity_cache_add(QD_ROUTER_LINK_TYPE, rlink);
- DEQ_INSERT_TAIL(router->links, rlink);
- qd_link_set_context(wp->out_link, rlink);
- qd_router_add_link_ref_LH(&addr->rlinks, rlink);
-
- if (DEQ_SIZE(addr->rlinks) == 1) {
- qd_field_iterator_t *iter = qd_address_iterator_string(wp->address, ITER_VIEW_ADDRESS_HASH);
- qd_address_iterator_set_phase(iter, wp->in_phase);
- //qd_router_mobile_added(router, iter);
- qd_field_iterator_free(iter);
- }
-
- pn_link_open(qd_link_pn(wp->out_link));
- qd_link_activate(wp->out_link);
-
- LOG(TRACE, "Sink out-link '%s'", pn_link_name(qd_link_pn(wp->out_link)));
- }
-}
-
-
-static void qd_waypoint_visit_source_LH(qd_dispatch_t *qd, qd_waypoint_t *wp)
-{
- qd_router_t *router = qd->router;
- qd_address_t *addr = wp->out_address;
- char unused;
-
- //
- // If the waypoint has no out-address, look it up in the hash table or create
- // a new one and put it in the hash table.
- //
- if (!addr) {
- //
- // Compose the phased-address and search the routing table for the address.
- // If it's not found, add it to the table but leave the link/router linkages empty.
- //
- qd_field_iterator_t *iter = qd_address_iterator_string(wp->address, ITER_VIEW_ADDRESS_HASH);
- qd_address_iterator_set_phase(iter, wp->out_phase);
- qd_hash_retrieve(router->addr_hash, iter, (void*) &addr);
-
- if (!addr) {
- addr = qd_address(router_semantics_for_addr(router, iter, wp->out_phase, &unused));
- qd_hash_insert(router->addr_hash, iter, addr, &addr->hash_handle);
- DEQ_INSERT_TAIL(router->addrs, addr);
- addr->waypoint = true;
- qd_entity_cache_add(QD_ROUTER_ADDRESS_TYPE, addr);
- }
-
- wp->out_address = addr;
- qd_field_iterator_free(iter);
- LOG(TRACE, "Source out-address=%s, out-phase=%c",
- qd_address_logstr(wp->out_address), wp->out_phase);
- }
-
- if (!wp->connected) {
- LOG(TRACE, "Source start connector %s", wp->connector_name);
- qd_connection_manager_start_on_demand(qd, wp->connector);
- }
- else if (!wp->in_link) {
- wp->in_link = qd_link(router->node, wp->connection, QD_INCOMING, wp->address);
- pn_terminus_set_address(qd_link_source(wp->in_link), wp->address);
-
- qd_router_link_t *rlink = 0; //qd_router_link(wp->in_link, QD_LINK_WAYPOINT, QD_INCOMING, addr, wp, 0);
-
- qd_entity_cache_add(QD_ROUTER_LINK_TYPE, rlink);
- DEQ_INSERT_TAIL(router->links, rlink);
- qd_link_set_context(wp->in_link, rlink);
-
- pn_link_open(qd_link_pn(wp->in_link));
- qd_link_activate(wp->in_link);
-
- LOG(TRACE, "Source in-link '%s'", pn_link_name(qd_link_pn(wp->in_link)));
- }
- if (wp->in_link && (DEQ_SIZE(addr->rlinks) + DEQ_SIZE(addr->rnodes) > 0)) {
- //
- // CASE: This address has reachable destinations in the network.
- // If there is no inbound link from the waypoint source,
- // establish one and issue credit.
- //
- pn_link_flow(qd_link_pn(wp->in_link), 1);
- qd_link_activate(wp->in_link);
- LOG(DEBUG, "Added credit for incoming link '%s'", pn_link_name(qd_link_pn(wp->in_link)));
- } else {
- //
- // CASE: This address has no reachable destinations in the network.
- //
- }
-}
-
-
-static void qd_waypoint_visit_LH(qd_dispatch_t *qd, qd_waypoint_t *wp)
-{
- if (wp->in_phase)
- qd_waypoint_visit_sink_LH(qd, wp);
- if (wp->out_phase)
- qd_waypoint_visit_source_LH(qd, wp);
-}
-
-
-static void qd_waypoint_activate_in_phase_LH(qd_dispatch_t *qd, qd_waypoint_t *wp)
-{
- //
- // Activate the "Sink" side of this waypoint.
- //
-
- //
- // If the in-phase is null, then no in-phase was configured for this waypoint.
- //
- if (wp->in_phase == '\0')
- return;
-
- //
- // Since the waypoint has an in-phase, we wish to advertise the waypoint's address
- // for senders to send to. We can't do this until a connection and link are set up
- // to the waypoint. Start the on-demand connector.
- //
- qd_connection_manager_start_on_demand(qd, wp->connector);
-}
-
-
-static void qd_waypoint_activate_out_phase_LH(qd_dispatch_t *qd, qd_waypoint_t *wp)
-{
- //
- // Activate the "Source" side of this waypoint.
- //
-
- // This function intentionally left blank
-}
-
-
-void qd_waypoint_activate_all(qd_dispatch_t *qd)
-{
- qd_router_t *router = qd->router;
- qd_waypoint_t *wp;
-
- sys_mutex_lock(router->lock);
- for (wp = DEQ_HEAD(router->waypoints); wp; wp = DEQ_NEXT(wp)) {
- //
- // Associate the waypoint with its named on-demand connector and for every
- // on-demand connector, create a list of associated waypoints.
- //
- if (!wp->connector) {
- wp->connector = qd_connection_manager_find_on_demand(qd, wp->connector_name);
- if (!wp->connector) {
- LOG(ERROR, "On-demand connector '%s' not found", wp->connector_name);
- continue;
- }
-
- qd_waypoint_context_t *context =
- (qd_waypoint_context_t*) qd_config_connector_context(wp->connector);
- if (!context) {
- context = NEW(qd_waypoint_context_t);
- DEQ_INIT(context->refs);
- qd_config_connector_set_context(wp->connector, context);
- }
- qd_waypoint_ref_t *ref = NEW(qd_waypoint_ref_t);
- DEQ_ITEM_INIT(ref);
- ref->wp = wp;
- DEQ_INSERT_TAIL(context->refs, ref);
- }
- }
-
- for (wp = DEQ_HEAD(router->waypoints); wp; wp = DEQ_NEXT(wp)) {
- qd_waypoint_activate_in_phase_LH(qd, wp);
- qd_waypoint_activate_out_phase_LH(qd, wp);
- qd_waypoint_visit_LH(qd, wp);
- }
- sys_mutex_unlock(router->lock);
-}
-
-
-void qd_waypoint_connection_opened(qd_dispatch_t *qd, qd_config_connector_t *cc, qd_connection_t *conn)
-{
- qd_waypoint_context_t *context = (qd_waypoint_context_t*) qd_config_connector_context(cc);
- if (!context)
- return;
-
- qd_log(qd->router->log_source, QD_LOG_INFO, "On-demand connector '%s' opened",
- qd_config_connector_name(cc));
-
- sys_mutex_lock(qd->router->lock);
- qd_waypoint_ref_t *ref = DEQ_HEAD(context->refs);
- while (ref) {
- ref->wp->connected = true;
- ref->wp->connection = conn;
- qd_waypoint_visit_LH(qd, ref->wp);
- ref = DEQ_NEXT(ref);
- }
- sys_mutex_unlock(qd->router->lock);
-}
-
-
-void qd_waypoint_new_incoming_link(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link)
-{
-}
-
-
-void qd_waypoint_new_outgoing_link(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link)
-{
-}
-
-
-void qd_waypoint_link_closed(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link)
-{
- if (qd_link_direction(link) == QD_INCOMING)
- wp->in_link = 0;
- else
- wp->out_link = 0;
-}
-
-
-void qd_waypoint_address_updated_LH(qd_dispatch_t *qd, qd_address_t *addr)
-{
- qd_waypoint_t *wp = DEQ_HEAD(qd->router->waypoints);
- while (wp) {
- LOG(TRACE, "Updated address %s", qd_address_logstr(addr));
- if (wp->out_address == addr)
- qd_waypoint_visit_LH(qd, wp);
- wp = DEQ_NEXT(wp);
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-dispatch/blob/2291e5cd/src/waypoint_private.h
----------------------------------------------------------------------
diff --git a/src/waypoint_private.h b/src/waypoint_private.h
deleted file mode 100644
index 4e064ea..0000000
--- a/src/waypoint_private.h
+++ /dev/null
@@ -1,61 +0,0 @@
-#ifndef __waypoint_private_h__
-#define __waypoint_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/server.h>
-#include <qpid/dispatch/connection_manager.h>
-#include "dispatch_private.h"
-
-/**
- * @file
- *
- * A waypoint sends/receives messages to/from an external entity such as a
- * broker as part of a multi-phase address.
- *
- * An address can have multiple phases. Each phase acts like a separate address,
- * but sharing the same address string.
- *
- * Phases are not visible to normal senders/receivers, they are set by
- * waypoints. Messages from normal senders go to the phase=0 address. Normal
- * subscribers subscribe to the highest phase defined for the address.
- *
- * A waypoint takes messages for its in-phase and sends them to the external
- * entity. Messages received from the external entity are given the waypoint's
- * out-phase. Waypoints can be "chained" with the out-phase of one equal to the
- * in-phase for the next. Thus waypoints provide a way to route messages via
- * multiple external entities between a sender and a subscriber using the same
- * address.
- */
-
-void qd_waypoint_activate_all(qd_dispatch_t *qd);
-
-/** Called when the router opens a connector associated with this waypoint */
-void qd_waypoint_connection_opened(qd_dispatch_t *qd, qd_config_connector_t *cc, qd_connection_t *conn);
-
-void qd_waypoint_new_incoming_link(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link);
-
-void qd_waypoint_new_outgoing_link(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link);
-
-void qd_waypoint_link_closed(qd_dispatch_t *qd, qd_waypoint_t *wp, qd_link_t *link);
-
-void qd_waypoint_address_updated_LH(qd_dispatch_t *qd, qd_address_t *address);
-
-
-#endif
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org