/*
 * 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 "router_core_private.h"
#include <qpid/dispatch/amqp.h>
#include <stdio.h>

//
// NOTE: If the in_delivery argument is NULL, the resulting out deliveries
//       shall be pre-settled.
//
typedef int (*qdr_forward_message_t) (qdr_core_t      *core,
                                      qdr_address_t   *addr,
                                      qd_message_t    *msg,
                                      qdr_delivery_t  *in_delivery,
                                      bool             exclude_inprocess,
                                      bool             control);

typedef bool (*qdr_forward_attach_t) (qdr_core_t     *core,
                                      qdr_address_t  *addr,
                                      qdr_link_t     *link,
                                      qdr_terminus_t *source,
                                      qdr_terminus_t *target);

struct qdr_forwarder_t {
    qdr_forward_message_t forward_message;
    qdr_forward_attach_t  forward_attach;
    bool                  bypass_valid_origins;
};

//==================================================================================
// Built-in Forwarders
//==================================================================================

static int qdr_forward_message_null_CT(qdr_core_t      *core,
                                       qdr_address_t   *addr,
                                       qd_message_t    *msg,
                                       qdr_delivery_t  *in_delivery,
                                       bool             exclude_inprocess,
                                       bool             control)
{
    qd_log(core->log, QD_LOG_CRITICAL, "NULL Message Forwarder Invoked");
    return 0;
}


static bool qdr_forward_attach_null_CT(qdr_core_t     *core,
                                       qdr_address_t  *addr,
                                       qdr_link_t     *link,
                                       qdr_terminus_t *source,
                                       qdr_terminus_t *target)
{
    qd_log(core->log, QD_LOG_CRITICAL, "NULL Attach Forwarder Invoked");
    return false;
}


static void qdr_forward_find_closest_remotes_CT(qdr_core_t *core, qdr_address_t *addr)
{
    qdr_node_t *rnode       = DEQ_HEAD(core->routers);
    int         lowest_cost = 0;

    if (!addr->closest_remotes)
        addr->closest_remotes = qd_bitmask(0);
    addr->cost_epoch  = core->cost_epoch;
    addr->next_remote = -1;

    qd_bitmask_clear_all(addr->closest_remotes);
    while (rnode) {
        if (qd_bitmask_value(addr->rnodes, rnode->mask_bit)) {
            if (lowest_cost == 0) {
                lowest_cost = rnode->cost;
                addr->next_remote = rnode->mask_bit;
            }
            if (lowest_cost == rnode->cost)
                qd_bitmask_set_bit(addr->closest_remotes, rnode->mask_bit);
            else
                break;
        }
        rnode = DEQ_NEXT(rnode);
    }
}


qdr_delivery_t *qdr_forward_new_delivery_CT(qdr_core_t *core, qdr_delivery_t *in_dlv, qdr_link_t *link, qd_message_t *msg)
{
    qdr_delivery_t *dlv = new_qdr_delivery_t();
    uint64_t       *tag = (uint64_t*) dlv->tag;

    ZERO(dlv);
    dlv->link    = link;
    dlv->msg     = qd_message_copy(msg);
    dlv->settled = !in_dlv || in_dlv->settled;
    *tag         = core->next_tag++;
    dlv->tag_length = 8;

    //
    // Create peer linkage only if the delivery is not settled
    //
    if (!dlv->settled) {
        if (in_dlv && in_dlv->peer == 0) {
            dlv->peer = in_dlv;
            in_dlv->peer = dlv;

            dlv->ref_count = 1;
            qdr_delivery_incref(in_dlv);
        }
    }

    return dlv;
}


void qdr_forward_deliver_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *dlv)
{
    sys_mutex_lock(link->conn->work_lock);
    DEQ_INSERT_TAIL(link->undelivered, dlv);
    dlv->where = QDR_DELIVERY_IN_UNDELIVERED;
    dlv->ref_count++; // We have the lock, don't use the incref function

    //
    // If the link isn't already on the links_with_deliveries list, put it there.
    //
    qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
    sys_mutex_unlock(link->conn->work_lock);

    //
    // Activate the outgoing connection for later processing.
    //
    qdr_connection_activate_CT(core, link->conn);
}


void qdr_forward_on_message(qdr_core_t *core, qdr_general_work_t *work)
{
    work->on_message(work->on_message_context, work->msg, work->maskbit, work->inter_router_cost);
    qd_message_free(work->msg);
}


void qdr_forward_on_message_CT(qdr_core_t *core, qdr_subscription_t *sub, qdr_link_t *link, qd_message_t *msg)
{
    qdr_general_work_t *work = qdr_general_work(qdr_forward_on_message);
    work->on_message         = sub->on_message;
    work->on_message_context = sub->on_message_context;
    work->msg                = qd_message_copy(msg);
    work->maskbit            = link ? link->conn->mask_bit : 0;
    work->inter_router_cost  = link ? link->conn->inter_router_cost : 1;
    qdr_post_general_work_CT(core, work);
}


int qdr_forward_multicast_CT(qdr_core_t      *core,
                             qdr_address_t   *addr,
                             qd_message_t    *msg,
                             qdr_delivery_t  *in_delivery,
                             bool             exclude_inprocess,
                             bool             control)
{
    bool          bypass_valid_origins = addr->forwarder->bypass_valid_origins;
    int           fanout               = 0;
    qd_bitmask_t *link_exclusion       = !!in_delivery ? in_delivery->link_exclusion : 0;
    bool          presettled           = !!in_delivery ? in_delivery->settled : true;

    //
    // If the delivery is not presettled, set the settled flag for forwarding so all
    // outgoing deliveries will be presettled.
    //
    // NOTE:  This is the only multicast mode currently supported.  Others will likely be
    //        implemented in the future.
    //
    if (!presettled)
        in_delivery->settled = true;

    //
    // Forward to local subscribers
    //
    if (!addr->local || exclude_inprocess) {
        qdr_link_ref_t *link_ref = DEQ_HEAD(addr->rlinks);
        while (link_ref) {
            qdr_link_t     *out_link     = link_ref->link;
            qdr_delivery_t *out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, out_link, msg);
            qdr_forward_deliver_CT(core, out_link, out_delivery);
            fanout++;
            if (out_link->link_type != QD_LINK_CONTROL && out_link->link_type != QD_LINK_ROUTER)
                addr->deliveries_egress++;
            link_ref = DEQ_NEXT(link_ref);
        }
    }

    //
    // Forward to remote routers with subscribers using the appropriate
    // link for the traffic class: control or data
    //
    //
    // 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;
    qd_field_iterator_t *ingress_iter = in_delivery ? in_delivery->origin : 0;

    if (ingress_iter && !bypass_valid_origins) {
        qd_address_iterator_reset_view(ingress_iter, ITER_VIEW_NODE_HASH);
        qdr_address_t *origin_addr;
        qd_hash_retrieve(core->addr_hash, ingress_iter, (void*) &origin_addr);
        if (origin_addr && qd_bitmask_cardinality(origin_addr->rnodes) == 1)
            qd_bitmask_first_set(origin_addr->rnodes, &origin);
    } else
        origin = 0;

    //
    // Forward to the next-hops for remote destinations.
    //
    if (origin >= 0) {
        int           dest_bit;
        qdr_link_t   *dest_link;
        qdr_node_t   *next_node;
        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.
        //
        int c;
        for (QD_BITMASK_EACH(addr->rnodes, dest_bit, c)) {
            qdr_node_t *rnode = core->routers_by_mask_bit[dest_bit];
            if (!rnode)
                continue;

            if (rnode->next_hop)
                next_node = rnode->next_hop;
            else
                next_node = rnode;

            dest_link = control ? next_node->peer_control_link : next_node->peer_data_link;
            if (dest_link && qd_bitmask_value(rnode->valid_origins, origin))
                qd_bitmask_set_bit(link_set, dest_link->conn->mask_bit);
        }

        //
        // 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 = control ?
                core->control_links_by_mask_bit[link_bit] :
                core->data_links_by_mask_bit[link_bit];
            if (dest_link && (!link_exclusion || qd_bitmask_value(link_exclusion, link_bit) == 0)) {
                qdr_delivery_t *out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, dest_link, msg);
                qdr_forward_deliver_CT(core, dest_link, out_delivery);
                fanout++;
                addr->deliveries_transit++;
            }
        }

        qd_bitmask_free(link_set);
    }

    if (!exclude_inprocess) {
        //
        // Forward to in-process subscribers
        //
        qdr_subscription_t *sub = DEQ_HEAD(addr->subscriptions);
        while (sub) {
            qdr_forward_on_message_CT(core, sub, in_delivery ? in_delivery->link : 0, msg);
            fanout++;
            addr->deliveries_to_container++;
            sub = DEQ_NEXT(sub);
        }
    }

    if (in_delivery && !presettled) {
        if (fanout == 0)
            //
            // The delivery was not presettled and it was not forwarded to any
            // destinations, return it to its original unsettled state.
            //
            in_delivery->settled = false;
        else {
            //
            // The delivery was not presettled and it was forwarded to at least
            // one destination.  Accept and settle the delivery.
            //
            in_delivery->disposition = PN_ACCEPTED;
            qdr_delivery_push_CT(core, in_delivery);
        }
    }

    return fanout;
}


int qdr_forward_closest_CT(qdr_core_t      *core,
                           qdr_address_t   *addr,
                           qd_message_t    *msg,
                           qdr_delivery_t  *in_delivery,
                           bool             exclude_inprocess,
                           bool             control)
{
    qdr_link_t     *out_link;
    qdr_delivery_t *out_delivery;

    //
    // Forward to an in-process subscriber if there is one.
    //
    if (!exclude_inprocess) {
        qdr_subscription_t *sub = DEQ_HEAD(addr->subscriptions);
        if (sub) {
            qdr_forward_on_message_CT(core, sub, in_delivery ? in_delivery->link : 0, msg);

            //
            // If the incoming delivery is not settled, it should be accepted and settled here.
            //
            if (in_delivery && !in_delivery->settled) {
                in_delivery->disposition = PN_ACCEPTED;
                in_delivery->settled     = true;
                qdr_delivery_push_CT(core, in_delivery);
            }

            //
            // Rotate this subscription to the end of the list to get round-robin distribution.
            //
            if (DEQ_SIZE(addr->subscriptions) > 1) {
                DEQ_REMOVE_HEAD(addr->subscriptions);
                DEQ_INSERT_TAIL(addr->subscriptions, sub);
            }

            addr->deliveries_to_container++;
            return 1;
        }
    }

    //
    // Forward to a local subscriber.
    //
    qdr_link_ref_t *link_ref = DEQ_HEAD(addr->rlinks);
    if (link_ref) {
        out_link     = link_ref->link;
        out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, out_link, msg);
        qdr_forward_deliver_CT(core, out_link, out_delivery);

        //
        // If there are multiple local subscribers, rotate the list of link references
        // so deliveries will be distributed among the subscribers in a round-robin pattern.
        //
        if (DEQ_SIZE(addr->rlinks) > 1) {
            DEQ_REMOVE_HEAD(addr->rlinks);
            DEQ_INSERT_TAIL(addr->rlinks, link_ref);
        }

        addr->deliveries_egress++;
        return 1;
    }

    //
    // If the cached list of closest remotes is stale (i.e. cost data has changed),
    // recompute the closest remote routers.
    //
    if (addr->cost_epoch != core->cost_epoch)
        qdr_forward_find_closest_remotes_CT(core, addr);

    //
    // Forward to remote routers with subscribers using the appropriate
    // link for the traffic class: control or data
    //
    qdr_node_t *next_node;

    if (addr->next_remote >= 0) {
        qdr_node_t *rnode = core->routers_by_mask_bit[addr->next_remote];
        if (rnode) {
            _qdbm_next(addr->closest_remotes, &addr->next_remote);
            if (addr->next_remote == -1)
                qd_bitmask_first_set(addr->closest_remotes, &addr->next_remote);

            if (rnode->next_hop)
                next_node = rnode->next_hop;
            else
                next_node = rnode;

            out_link = control ? next_node->peer_control_link : next_node->peer_data_link;
            if (out_link) {
                out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, out_link, msg);
                qdr_forward_deliver_CT(core, out_link, out_delivery);
                addr->deliveries_transit++;
                return 1;
            }
        }
    }

    return 0;
}


int qdr_forward_balanced_CT(qdr_core_t      *core,
                            qdr_address_t   *addr,
                            qd_message_t    *msg,
                            qdr_delivery_t  *in_delivery,
                            bool             exclude_inprocess,
                            bool             control)
{
    //
    // Control messages should never use balanced treatment.
    //
    assert(!control);

    //
    // If this is the first time through here, allocate the array for outstanding delivery counts.
    //
    if (addr->outstanding_deliveries == 0) {
        addr->outstanding_deliveries = NEW_ARRAY(int, qd_bitmask_width());
        for (int i = 0; i < qd_bitmask_width(); i++)
            addr->outstanding_deliveries[i] = 0;
    }

    qdr_link_t *best_eligible_link       = 0;
    int         best_eligible_link_bit   = -1;
    uint32_t    eligible_link_value      = UINT32_MAX;
    qdr_link_t *best_ineligible_link     = 0;
    int         best_ineligible_link_bit = -1;
    uint32_t    ineligible_link_value    = UINT32_MAX;

    //
    // Find all the possible outbound links for this delivery, searching for the one with the
    // smallest eligible value.  Value = outstanding_deliveries + minimum_downrange_cost.
    // A link is ineligible if the outstanding_deliveries is equal to or greater than the
    // link's capacity.
    //
    // If there are no eligible links, use the best ineligible link.  Zero fanout should be returned
    // only if there are no available destinations.
    //

    //
    // Start with the local links
    //
    qdr_link_ref_t *link_ref = DEQ_HEAD(addr->rlinks);
    while (link_ref) {
        qdr_link_t *link     = link_ref->link;
        uint32_t    value    = DEQ_SIZE(link->undelivered) + DEQ_SIZE(link->unsettled);
        bool        eligible = link->capacity > value;

        //
        // If this is the best eligible link so far, record the fact.
        // Otherwise, if this is the best ineligible link, make note of that.
        //
        if (eligible && eligible_link_value > value) {
            best_eligible_link  = link;
            eligible_link_value = value;
        } else if (!eligible && ineligible_link_value > value) {
            best_ineligible_link  = link;
            ineligible_link_value = value;
        }

        link_ref = DEQ_NEXT(link_ref);
    }

    //
    // If we haven't already found a link with zero (best possible) value, check the
    // inter-router links as well.
    //
    if (!best_eligible_link || eligible_link_value > 0) {
        //
        // 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 = 0;
        qd_field_iterator_t *ingress_iter = in_delivery ? in_delivery->origin : 0;

        if (ingress_iter) {
            qd_address_iterator_reset_view(ingress_iter, ITER_VIEW_NODE_HASH);
            qdr_address_t *origin_addr;
            qd_hash_retrieve(core->addr_hash, ingress_iter, (void*) &origin_addr);
            if (origin_addr && qd_bitmask_cardinality(origin_addr->rnodes) == 1)
                qd_bitmask_first_set(origin_addr->rnodes, &origin);
        }

        int c;
        int node_bit;
        for (QD_BITMASK_EACH(addr->rnodes, node_bit, c)) {
            qdr_node_t *rnode     = core->routers_by_mask_bit[node_bit];
            qdr_node_t *next_node = rnode->next_hop ? rnode->next_hop : rnode;
            qdr_link_t *link      = next_node->peer_data_link;
            if (!link) continue;
            int         link_bit  = link->conn->mask_bit;
            int         value     = addr->outstanding_deliveries[link_bit];
            bool        eligible  = link->capacity > value;

            if (qd_bitmask_value(rnode->valid_origins, origin)) {
                //
                // Link is a candidate, adjust the value by the bias (node cost).
                //
                value += rnode->cost;
                if (eligible && eligible_link_value > value) {
                    best_eligible_link     = link;
                    best_eligible_link_bit = link_bit;
                    eligible_link_value    = value;
                } else if (!eligible && ineligible_link_value > value) {
                    best_ineligible_link     = link;
                    best_ineligible_link_bit = link_bit;
                    ineligible_link_value    = value;
                }
            }
        }
    } else if (best_eligible_link) {
        //
        // Rotate the rlinks list to enhance the appearance of balance when there is
        // little load (see DISPATCH-367)
        //
        if (DEQ_SIZE(addr->rlinks) > 1) {
            link_ref = DEQ_HEAD(addr->rlinks);
            DEQ_REMOVE_HEAD(addr->rlinks);
            DEQ_INSERT_TAIL(addr->rlinks, link_ref);
        }
    }

    qdr_link_t *chosen_link     = 0;
    int         chosen_link_bit = -1;

    if (best_eligible_link) {
        chosen_link     = best_eligible_link;
        chosen_link_bit = best_eligible_link_bit;
    } else if (best_ineligible_link) {
        chosen_link     = best_ineligible_link;
        chosen_link_bit = best_ineligible_link_bit;
    }

    if (chosen_link) {
        qdr_delivery_t *out_delivery = qdr_forward_new_delivery_CT(core, in_delivery, chosen_link, msg);
        qdr_forward_deliver_CT(core, chosen_link, out_delivery);

        //
        // If the delivery is unsettled and the link is inter-router, account for the outstanding delivery.
        //
        if (in_delivery && !in_delivery->settled && chosen_link_bit >= 0) {
            addr->outstanding_deliveries[chosen_link_bit]++;
            out_delivery->tracking_addr = addr;
            addr->tracked_deliveries++;
        }

        //
        // Bump the appropriate counter based on where we sent the delivery.
        //
        if (chosen_link_bit >= 0)
            addr->deliveries_transit++;
        else
            addr->deliveries_egress++;
        return 1;
    }

    return 0;
}


bool qdr_forward_link_balanced_CT(qdr_core_t     *core,
                                  qdr_address_t  *addr,
                                  qdr_link_t     *in_link,
                                  qdr_terminus_t *source,
                                  qdr_terminus_t *target)
{
    qdr_connection_ref_t *conn_ref = DEQ_HEAD(addr->conns);
    qdr_connection_t     *conn     = 0;

    //
    // Check for locally connected containers that can handle this link attach.
    //
    if (conn_ref) {
        conn = conn_ref->conn;

        //
        // If there are more than one local connections available for handling this link,
        // rotate the list so the attaches are balanced across the containers.
        //
        if (DEQ_SIZE(addr->conns) > 1) {
            DEQ_REMOVE_HEAD(addr->conns);
            DEQ_INSERT_TAIL(addr->conns, conn_ref);
        }
    } else {
        //
        // Look for a next-hop we can use to forward the link-attach.
        //
        qdr_node_t *next_node;

        if (addr->cost_epoch != core->cost_epoch) {
            addr->next_remote = -1;
            addr->cost_epoch  = core->cost_epoch;
        }

        if (addr->next_remote < 0) {
            qd_bitmask_first_set(addr->rnodes, &addr->next_remote);
        }

        if (addr->next_remote >= 0) {

            qdr_node_t *rnode = core->routers_by_mask_bit[addr->next_remote];

            if (rnode) {
                //
                // Advance the addr->next_remote so there will be link balance across containers
                //
                _qdbm_next(addr->rnodes, &addr->next_remote);
                if (addr->next_remote == -1)
                    qd_bitmask_first_set(addr->rnodes, &addr->next_remote);

                if (rnode->next_hop)
                    next_node = rnode->next_hop;
                else
                    next_node = rnode;

                if (next_node && next_node->peer_data_link)
                    conn = next_node->peer_data_link->conn;
            }
        }
    }

    if (conn) {
        qdr_link_t *out_link = new_qdr_link_t();
        ZERO(out_link);
        out_link->core           = core;
        out_link->identity       = qdr_identifier(core);
        out_link->conn           = conn;
        out_link->link_type      = QD_LINK_ENDPOINT;
        out_link->link_direction = qdr_link_direction(in_link) == QD_OUTGOING ? QD_INCOMING : QD_OUTGOING;
        out_link->admin_enabled  = true;
        out_link->oper_status    = QDR_LINK_OPER_DOWN;

        out_link->name = (char*) malloc(strlen(in_link->name) + 1);
        strcpy(out_link->name, in_link->name);

        out_link->connected_link = in_link;
        in_link->connected_link  = out_link;

        DEQ_INSERT_TAIL(core->open_links, out_link);
        qdr_add_link_ref(&conn->links, out_link, QDR_LINK_LIST_CLASS_CONNECTION);

        qdr_connection_work_t *work = new_qdr_connection_work_t();
        ZERO(work);
        work->work_type = QDR_CONNECTION_WORK_FIRST_ATTACH;
        work->link      = out_link;
        work->source    = source;
        work->target    = target;

        qdr_connection_enqueue_work_CT(core, conn, work);

        return true;
    }

    return false;
}


//==================================================================================
// In-Thread API Functions
//==================================================================================

qdr_forwarder_t *qdr_new_forwarder(qdr_forward_message_t fm, qdr_forward_attach_t fa, bool bypass_valid_origins)
{
    qdr_forwarder_t *forw = NEW(qdr_forwarder_t);

    forw->forward_message      = fm ? fm : qdr_forward_message_null_CT;
    forw->forward_attach       = fa ? fa : qdr_forward_attach_null_CT;
    forw->bypass_valid_origins = bypass_valid_origins;

    return forw;
}


void qdr_forwarder_setup_CT(qdr_core_t *core)
{
    //
    // Create message forwarders
    //
    core->forwarders[QD_TREATMENT_MULTICAST_FLOOD]  = qdr_new_forwarder(qdr_forward_multicast_CT, 0, true);
    core->forwarders[QD_TREATMENT_MULTICAST_ONCE]   = qdr_new_forwarder(qdr_forward_multicast_CT, 0, false);
    core->forwarders[QD_TREATMENT_ANYCAST_CLOSEST]  = qdr_new_forwarder(qdr_forward_closest_CT,   0, false);
    core->forwarders[QD_TREATMENT_ANYCAST_BALANCED] = qdr_new_forwarder(qdr_forward_balanced_CT,  0, false);

    //
    // Create link forwarders
    //
    core->forwarders[QD_TREATMENT_LINK_BALANCED] = qdr_new_forwarder(0, qdr_forward_link_balanced_CT, false);
}


qdr_forwarder_t *qdr_forwarder_CT(qdr_core_t *core, qd_address_treatment_t treatment)
{
    if (treatment <= QD_TREATMENT_LINK_BALANCED)
        return core->forwarders[treatment];
    return 0;
}


int qdr_forward_message_CT(qdr_core_t *core, qdr_address_t *addr, qd_message_t *msg, qdr_delivery_t *in_delivery,
                           bool exclude_inprocess, bool control)
{
    if (addr->forwarder)
        return addr->forwarder->forward_message(core, addr, msg, in_delivery, exclude_inprocess, control);

    // TODO - Deal with this delivery's disposition
    return 0;
}


bool qdr_forward_attach_CT(qdr_core_t *core, qdr_address_t *addr, qdr_link_t *in_link,
                           qdr_terminus_t *source, qdr_terminus_t *target)
{
    if (addr->forwarder)
        return addr->forwarder->forward_attach(core, addr, in_link, source, target);
    return false;
}