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


//==================================================================================
// Internal Functions
//==================================================================================

static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard);


//==================================================================================
// Interface Functions
//==================================================================================

qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_iterator_t *ingress,
                                 bool settled, qd_bitmask_t *link_exclusion, int ingress_index)
{
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    set_safe_ptr_qdr_link_t(link, &dlv->link_sp);
    dlv->msg            = msg;
    dlv->to_addr        = 0;
    dlv->origin         = ingress;
    dlv->settled        = settled;
    dlv->presettled     = settled;
    dlv->link_exclusion = link_exclusion;
    dlv->ingress_index  = ingress_index;
    dlv->error          = 0;
    dlv->disposition    = 0;

    qdr_delivery_incref(dlv, "qdr_link_deliver - newly created delivery, add to action list");
    qdr_delivery_incref(dlv, "qdr_link_deliver - protect returned value");

    action->args.delivery.delivery = dlv;
    action->args.delivery.more = !qd_message_receive_complete(msg);
    qdr_action_enqueue(link->core, action);
    return dlv;
}


qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
                                    qd_iterator_t *ingress, qd_iterator_t *addr,
                                    bool settled, qd_bitmask_t *link_exclusion, int ingress_index)
{
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    set_safe_ptr_qdr_link_t(link, &dlv->link_sp);
    dlv->msg            = msg;
    dlv->to_addr        = addr;
    dlv->origin         = ingress;
    dlv->settled        = settled;
    dlv->presettled     = settled;
    dlv->link_exclusion = link_exclusion;
    dlv->ingress_index  = ingress_index;
    dlv->error          = 0;
    dlv->disposition    = 0;

    qdr_delivery_incref(dlv, "qdr_link_deliver_to - newly created delivery, add to action list");
    qdr_delivery_incref(dlv, "qdr_link_deliver_to - protect returned value");

    action->args.delivery.delivery = dlv;
    action->args.delivery.more = !qd_message_receive_complete(msg);
    qdr_action_enqueue(link->core, action);
    return dlv;
}


qdr_delivery_t *qdr_link_deliver_to_routed_link(qdr_link_t *link, qd_message_t *msg, bool settled,
                                                const uint8_t *tag, int tag_length,
                                                uint64_t disposition, pn_data_t* disposition_data)
{
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    set_safe_ptr_qdr_link_t(link, &dlv->link_sp);
    dlv->msg          = msg;
    dlv->settled      = settled;
    dlv->presettled   = settled;
    dlv->error        = 0;
    dlv->disposition  = 0;

    qdr_delivery_set_extension_state(dlv, disposition, disposition_data, true);
    qdr_delivery_incref(dlv, "qdr_link_deliver_to_routed_link - newly created delivery, add to action list");
    qdr_delivery_incref(dlv, "qdr_link_deliver_to_routed_link - protect returned value");

    action->args.delivery.delivery = dlv;
    action->args.delivery.more = !qd_message_receive_complete(msg);
    action->args.delivery.tag_length = tag_length;
    assert(tag_length <= QDR_DELIVERY_TAG_MAX);
    memcpy(action->args.delivery.tag, tag, tag_length);
    qdr_action_enqueue(link->core, action);
    return dlv;
}


// send up to credit pending outgoing deliveries
int qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
{
    qdr_connection_t *conn = link->conn;
    qdr_delivery_t   *dlv;
    int               offer   = -1;
    bool              settled = false;
    bool              send_complete = false;
    int               num_deliveries_completed = 0;

    if (link->link_direction == QD_OUTGOING) {

        // If a detach has been received on the link, there is no need to process deliveries on the link.
        if (link->detach_received)
            return 0;

        while (credit > 0) {
            sys_mutex_lock(conn->work_lock);
            dlv = DEQ_HEAD(link->undelivered);
            if (dlv) {
                qdr_delivery_incref(dlv, "qdr_link_process_deliveries - holding the undelivered delivery locally");
                uint64_t new_disp = 0;

                // DISPATCH-1302 race hack fix: There is a race between the CORE thread
                // and the outbound (this) thread over settlement. It occurs when the CORE
                // thread is trying to propagate settlement to a peer (this delivery)
                // while this thread is in core->deliver_handler.  This can result in the
                // CORE thread NOT pushing the peer delivery change since it is not yet off of
                // the undelivered list, while this thread does not settle because it missed
                // the settled flag update.
                do {
                    settled = dlv->settled;
                    sys_mutex_unlock(conn->work_lock);
                    new_disp = core->deliver_handler(core->user_context, link, dlv, settled);
                    sys_mutex_lock(conn->work_lock);
                } while (settled != dlv->settled);  // oops missed the settlement
                send_complete = qdr_delivery_send_complete(dlv);
                if (send_complete) {
                    //
                    // The entire message has been sent. It is now the appropriate time to have the delivery removed
                    // from the head of the undelivered list and move it to the unsettled list if it is not settled.
                    //
                    num_deliveries_completed++;

                    credit--;
                    link->credit_to_core--;
                    link->total_deliveries++;

                    // DISPATCH-1153:
                    // If the undelivered list is cleared the link may have detached.  Stop processing.
                    offer = DEQ_SIZE(link->undelivered);
                    if (offer == 0) {
                        qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - release local reference - closed link");
                        sys_mutex_unlock(conn->work_lock);
                        return num_deliveries_completed;
                    }

                    assert(dlv == DEQ_HEAD(link->undelivered));
                    DEQ_REMOVE_HEAD(link->undelivered);
                    dlv->link_work = 0;

                    if (settled) {
                        dlv->where = QDR_DELIVERY_NOWHERE;
                        qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - remove from undelivered list");
                    } else {
                        DEQ_INSERT_TAIL(link->unsettled, dlv);
                        dlv->where = QDR_DELIVERY_IN_UNSETTLED;
                        qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer:  dlv:%lx qdr_link_process_deliveries: undelivered-list -> unsettled-list", (long) dlv);
                    }
                }
                else {
                    qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - release local reference - not send_complete");

                    //
                    // The message is still being received/sent.
                    // 1. We cannot remove the delivery from the undelivered list.
                    //    This delivery needs to stay at the head of the undelivered list until the entire message
                    //    has been sent out i.e other deliveries in the undelivered list have to wait before this
                    //    entire large delivery is sent out
                    // 2. We need to call deliver_handler so any newly arrived bytes can be pushed out
                    // 3. We need to break out of this loop otherwise a thread will keep spinning in here until
                    //    the entire message has been sent out.
                    //
                    sys_mutex_unlock(conn->work_lock);

                    //
                    // Note here that we are not incrementing num_deliveries_processed. Since this delivery is
                    // still coming in or still being sent out, we cannot consider this delivery as fully processed.
                    //
                    return num_deliveries_completed;
                }
                sys_mutex_unlock(conn->work_lock);

                if (new_disp) {
                    // the remote sender-settle-mode forced us to pre-settle the
                    // message.  The core needs to know this, so we "fake" receiving a
                    // settle+disposition update from the remote end of the link:
                    qdr_delivery_remote_state_updated(core, dlv, new_disp, true, 0, 0, false);
                }

                qdr_delivery_decref(core, dlv, "qdr_link_process_deliveries - release local reference - done processing");
            } else {
                sys_mutex_unlock(conn->work_lock);
                break;
            }
        }

        if (offer != -1)
            core->offer_handler(core->user_context, link, offer);
    }

    return num_deliveries_completed;
}


void qdr_link_flow(qdr_core_t *core, qdr_link_t *link, int credit, bool drain_mode)
{
    qdr_action_t *action = qdr_action(qdr_link_flow_CT, "link_flow");

    //
    // Compute the number of credits now available that we haven't yet given
    // incrementally to the router core.  i.e. convert absolute credit to
    // incremental credit.
    //
    if (link->drain_mode && !drain_mode) {
        link->credit_to_core = 0;   // credit calc reset when coming out of drain mode
    } else {
        credit -= link->credit_to_core;
        if (credit < 0)
            credit = 0;
        link->credit_to_core += credit;
    }

    set_safe_ptr_qdr_link_t(link, &action->args.connection.link);
    action->args.connection.credit = credit;
    action->args.connection.drain  = drain_mode;

    qdr_action_enqueue(core, action);
    qdr_record_link_credit(core, link);
}

void qdr_link_set_drained(qdr_core_t *core, qdr_link_t *link)
{
    if (link) {
        link->drain_mode = false;
        link->credit_to_core = 0;
    }
}


void qdr_send_to1(qdr_core_t *core, qd_message_t *msg, qd_iterator_t *addr, bool exclude_inprocess, bool control)
{
    qdr_action_t *action = qdr_action(qdr_send_to_CT, "send_to");
    action->args.io.address           = qdr_field_from_iter(addr);
    action->args.io.message           = qd_message_copy(msg);
    action->args.io.exclude_inprocess = exclude_inprocess;
    action->args.io.control           = control;

    qdr_action_enqueue(core, action);
}


void qdr_send_to2(qdr_core_t *core, qd_message_t *msg, const char *addr, bool exclude_inprocess, bool control)
{
    qdr_action_t *action = qdr_action(qdr_send_to_CT, "send_to");
    action->args.io.address           = qdr_field(addr);
    action->args.io.message           = qd_message_copy(msg);
    action->args.io.exclude_inprocess = exclude_inprocess;
    action->args.io.control           = control;

    qdr_action_enqueue(core, action);
}


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


static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    qdr_link_t *link = safe_deref_qdr_link_t(action->args.connection.link);

    if (discard || !link)
        return;

    int  credit           = action->args.connection.credit;
    bool drain            = action->args.connection.drain;
    bool activate         = false;
    bool drain_was_set    = !link->drain_mode && drain;
    qdr_link_work_t *work = 0;

    link->drain_mode = drain;

    //
    // If the link was stalled due to internal backpressure from the transport, put it
    // on the links-with-work list and activate the connection to resume sending.
    //
    if (link->stalled_outbound) {
        link->stalled_outbound = false;

        sys_mutex_lock(link->conn->work_lock);

        if (DEQ_SIZE(link->undelivered) > 0) {
            qdr_add_link_ref(&link->conn->links_with_work[link->priority], link, QDR_LINK_LIST_CLASS_WORK);
            activate = true;
        }

        sys_mutex_unlock(link->conn->work_lock);
    }

    if (link->core_endpoint) {
        qdrc_endpoint_do_flow_CT(core, link->core_endpoint, credit, drain);
    } else if (link->connected_link) {
        //
        // If this is an attach-routed link, propagate the flow data downrange.
        // Note that the credit value is incremental.
        //
        qdr_link_t *clink = link->connected_link;

        if (clink->link_direction == QD_INCOMING)
            qdr_link_issue_credit_CT(core, link->connected_link, credit, drain);
        else {
            work = new_qdr_link_work_t();
            ZERO(work);
            work->work_type = QDR_LINK_WORK_FLOW;
            work->value     = credit;
            if (drain)
                work->drain_action = QDR_LINK_WORK_DRAIN_ACTION_DRAINED;
            qdr_link_enqueue_work_CT(core, clink, work);
        }
    } else {
        if (link->attach_count == 1)
            //
            // The link is half-open.  Store the pending credit to be dealt with once the link is
            // progressed to the next step.
            //
            link->credit_stored += credit;

        //
        // Handle the replenishing of credit outbound
        //
        if (link->link_direction == QD_OUTGOING && (credit > 0 || drain_was_set)) {
            if (drain_was_set) {
                work = new_qdr_link_work_t();
                ZERO(work);
                work->work_type    = QDR_LINK_WORK_FLOW;
                work->drain_action = QDR_LINK_WORK_DRAIN_ACTION_DRAINED;
            }

            sys_mutex_lock(link->conn->work_lock);
            if (work)
                DEQ_INSERT_TAIL(link->work_list, work);
            if (DEQ_SIZE(link->undelivered) > 0 || drain_was_set) {
                qdr_add_link_ref(&link->conn->links_with_work[link->priority], link, QDR_LINK_LIST_CLASS_WORK);
                activate = true;
            }
            sys_mutex_unlock(link->conn->work_lock);
        } else if (link->link_direction == QD_INCOMING) {
            if (drain) {
                link->credit_pending = link->capacity;
            }
        }
    }

    //
    // Activate the connection if we have deliveries to send or drain mode was set.
    //
    if (activate)
        qdr_connection_activate_CT(core, link->conn);
}


/**
 * Return the number of outbound paths to destinations that this address has.
 * Note that even if there are more than zero paths, the destination still may
 * be unreachable (e.g. an rnode next hop with no link).
 */
static long qdr_addr_path_count_CT(qdr_address_t *addr)
{
    if (!addr)
        return 0;

    long rc = ((long) DEQ_SIZE(addr->subscriptions)
               + (long) DEQ_SIZE(addr->rlinks)
               + (long) qd_bitmask_cardinality(addr->rnodes));
    if (addr->exchange)
        rc += qdr_exchange_binding_count(addr->exchange)
            + ((qdr_exchange_alternate_addr(addr->exchange)) ? 1 : 0);
    return rc;
}


static void qdr_link_forward_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *dlv, qdr_address_t *addr, bool more)
{
    qdr_link_t *dlv_link = qdr_delivery_link(dlv);

    assert(dlv_link == link);

    if (!dlv_link)
        return;

    if (dlv_link->link_type == QD_LINK_ENDPOINT && !dlv_link->fallback)
        core->deliveries_ingress++;

    if (addr
        && addr == link->owning_addr
        && qdr_addr_path_count_CT(addr) == 0
        && (link->fallback || qdr_addr_path_count_CT(addr->fallback) == 0)) {
        //
        // We are trying to forward a delivery on an address that has no outbound paths
        // AND the incoming link is targeted (not anonymous).
        //
        // We shall release the delivery (it is currently undeliverable). Since
        // there are no receivers we will try to drain credit to prevent the
        // sender from attempting to send more to this address.
        //
        if (dlv->settled) {
            // Increment the presettled_dropped_deliveries on the in_link
            link->dropped_presettled_deliveries++;
            if (dlv_link->link_type == QD_LINK_ENDPOINT)
                core->dropped_presettled_deliveries++;
        }

        //
        // Note if the message was pre-settled we still call the
        // qdr_delivery_release_CT so if this delivery is multi-frame we can
        // restart receiving the delivery in case it is stalled. Note that
        // messages will not *actually* be released in this case because these
        // are presettled messages.
        //
        qdr_delivery_release_CT(core, dlv);

        //
        // Credit update: since this is a targeted link to an address for which
        // there is no consumers then do not replenish credit - drain instead.
        // However edge is a special snowflake which always has credit available.
        //
        if (link->edge) {
            qdr_link_issue_credit_CT(core, link, 1, false);
        } else {
            qdr_link_issue_credit_CT(core, link, 0, true);  // drain
            link->credit_pending++;
        }

        qdr_delivery_decref_CT(core, dlv, "qdr_link_forward_CT - removed from action (no path)");
        return;
    }

    int fanout = 0;

    dlv->multicast = qdr_is_addr_treatment_multicast(addr);

    if (addr) {
        fanout = qdr_forward_message_CT(core, addr, dlv->msg, dlv, false, link->link_type == QD_LINK_CONTROL);
        if (link->link_type != QD_LINK_CONTROL && link->link_type != QD_LINK_ROUTER) {
            if (!link->fallback)
                addr->deliveries_ingress++;

            if (qdr_connection_route_container(link->conn)) {
                addr->deliveries_ingress_route_container++;
                core->deliveries_ingress_route_container++;
            }

        }
    } else {
        //
        // There is no address that we can send this delivery to, which means
        // the addr was not found in our hash table. This can be because there
        // were no receivers or because the address was not defined in the
        // config file.
        //

        qd_address_treatment_t trt = core->qd->default_treatment;
        if (dlv->to_addr) {
            qdr_address_config_t *ignore = 0;
            trt = qdr_treatment_for_address_hash_with_default_CT(core,
                                                                 dlv->to_addr,
                                                                 trt,
                                                                 &ignore);
        }

        if (trt == QD_TREATMENT_UNAVAILABLE) {
            //
            // The treatment for these addresses is set to be unavailable, we
            // stop trying to forward it.  If the link is a locally attached client
            // we reject the message if the link is not anonymous as per the
            // documentation of the router's defaultTreatment=unavailable.  We
            // simply release it for other link types as the message did have a
            // destination at some point (it was forwarded to this router after
            // all) - the loss of the destination may be temporary.
            //
            if (link->link_type == QD_LINK_ENDPOINT) {
                dlv->error = qdr_error(QD_AMQP_COND_NOT_FOUND, "Deliveries cannot be sent to an unavailable address");
                qdr_delivery_reject_CT(core, dlv);
                if (qdr_link_is_anonymous(link)) {
                    qdr_link_issue_credit_CT(core, link, 1, false);
                } else {
                    // cannot forward on this targeted link.  withhold credit and drain
                    qdr_link_issue_credit_CT(core, link, 0, true);
                }
            } else {
                qdr_delivery_release_CT(core, dlv);
                qdr_link_issue_credit_CT(core, link, 1, false);
            }
            //
            // We will not detach this link because this could be anonymous sender. We don't know
            // which address the sender will be sending to next
            // If this was not an anonymous sender, the initial attach would have been rejected if the target address was unavailable.
            //
            qdr_delivery_decref_CT(core, dlv, "qdr_link_forward_CT - removed from action (treatment unavailable)");
            return;
        }
    }

    //
    // If the anonymous delivery could not be sent anywhere (fanout = 0) and it is not multicasted, try sending it over
    // the anonymous link.
    //
    if (fanout == 0 && !dlv->multicast && link->owning_addr == 0 && dlv->to_addr != 0) {
        if (core->edge_conn_addr && link->conn->role != QDR_ROLE_EDGE_CONNECTION) {
            qdr_address_t *sender_address = core->edge_conn_addr(core->edge_context);
            if (sender_address && sender_address != addr)
                fanout += qdr_forward_message_CT(core, sender_address, dlv->msg, dlv, false, link->link_type == QD_LINK_CONTROL);
        }
    }

    //
    // If the fanout is still zero, check to see if there is a fallback address and
    // route via the fallback if present.  Don't do fallback forwarding if this link is
    // itself associated with a fallback destination.
    //
    if (fanout == 0 && !!addr && !!addr->fallback && !link->fallback) {
        const char          *key      = (const char*) qd_hash_key_by_handle(addr->fallback->hash_handle);
        qd_composed_field_t *to_field = qd_compose_subfield(0);
        qd_compose_insert_string(to_field, key + 2);
        qd_message_set_to_override_annotation(dlv->msg, to_field);
        qd_message_set_phase_annotation(dlv->msg, key[1] - '0');
        fanout = qdr_forward_message_CT(core, addr->fallback, dlv->msg, dlv, false, link->link_type == QD_LINK_CONTROL);
        if (fanout > 0) {
            addr->deliveries_redirected++;
            core->deliveries_redirected++;
        }
    }

    if (fanout == 0) {
        //
        // Message was not delivered, drop the delivery.
        //
        // If the delivery is not settled, release it.
        //
        if (!dlv->settled)
            qdr_delivery_release_CT(core, dlv);
        else {
            link->dropped_presettled_deliveries++;
            if (dlv_link->link_type == QD_LINK_ENDPOINT)
                core->dropped_presettled_deliveries++;
        }

        //
        // Decrementing the delivery ref count for the action
        //
        qdr_delivery_decref_CT(core, dlv, "qdr_link_forward_CT - removed from action (1)");
        qdr_link_issue_credit_CT(core, link, 1, false);
    } else if (fanout > 0) {
        if (dlv->settled) {
            //
            // The delivery is settled.  Keep it off the unsettled list and issue
            // replacement credit for it now.
            //
            qdr_link_issue_credit_CT(core, link, 1, false);
            if (!more) {
                //
                // This decref is for the action ref
                //
                qdr_delivery_decref_CT(core, dlv, "qdr_link_forward_CT - removed from action (2)");
            }
            else {
                //
                // The message is still coming through since receive_complete is false. We have to put this delivery in the settled list.
                // We need to do this because we have linked this delivery to a peer.
                // If this connection goes down, we will have to unlink peer so that peer knows that its peer is not-existent anymore
                // and need to tell the other side that the message has been aborted.
                //

                //
                // Again, don't bother decrementing then incrementing the ref_count, we are still using the action ref count
                //
                DEQ_INSERT_TAIL(link->settled, dlv);
                dlv->where = QDR_DELIVERY_IN_SETTLED;
                qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer:  dlv:%lx qdr_link_forward_CT: action-list -> settled-list", (long) dlv);
            }
        } else {
            //
            // Again, don't bother decrementing then incrementing the ref_count
            //
            DEQ_INSERT_TAIL(link->unsettled, dlv);
            dlv->where = QDR_DELIVERY_IN_UNSETTLED;
            qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer:  dlv:%lx qdr_link_forward_CT: action-list -> unsettled-list", (long) dlv);

            //
            // If the delivery was received on an inter-router link, issue the credit
            // now.  We don't want to tie inter-router link flow control to unsettled
            // deliveries because it increases the risk of credit starvation if there
            // are many addresses sharing the link.
            //
            if (link->link_type == QD_LINK_CONTROL || link->link_type == QD_LINK_ROUTER || link->edge)
                qdr_link_issue_credit_CT(core, link, 1, false);
        }
    }
}


static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    if (discard)
        return;

    qdr_delivery_t *dlv  = action->args.delivery.delivery;
    bool            more = action->args.delivery.more;
    qdr_link_t     *link = qdr_delivery_link(dlv);

    if (!link)
        return;
    if (link->conn)
        link->conn->last_delivery_time = core->uptime_ticks;

    link->total_deliveries++;

    //
    // Record the ingress time so we can track the age of this delivery.
    //
    dlv->ingress_time = core->uptime_ticks;

    //
    // If the link is an edge link, mark this delivery as via-edge
    //
    dlv->via_edge = link->edge;

    //
    // If this link has a core_endpoint, direct deliveries to that endpoint.
    //
    if (!!link->core_endpoint) {
        qdrc_endpoint_do_deliver_CT(core, link->core_endpoint, dlv);
        return;
    }

    if (link->connected_link) {
        if (link->link_direction == QD_INCOMING)
            core->deliveries_ingress++;

        //
        // If this is an attach-routed link, put the delivery directly onto the peer link
        //
        qdr_delivery_t *peer = qdr_forward_new_delivery_CT(core, dlv, link->connected_link, dlv->msg);
        qdr_delivery_set_extension_state(peer,
                                         dlv->remote_disposition,
                                         qdr_delivery_extension_state(dlv),
                                         true);

        //
        // Copy the delivery tag.  For link-routing, the delivery tag must be preserved.
        //
        peer->tag_length = action->args.delivery.tag_length;
        memcpy(peer->tag, action->args.delivery.tag, peer->tag_length);

        qdr_forward_deliver_CT(core, link->connected_link, peer);

        if (!dlv->settled) {
            DEQ_INSERT_TAIL(link->unsettled, dlv);
            dlv->where = QDR_DELIVERY_IN_UNSETTLED;
            qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer:  dlv:%lx qdr_link_deliver_CT: action-list -> unsettled-list", (long) dlv);
        } else {
            //
            // If the delivery is settled, decrement the ref_count on the delivery.
            // This count was the owned-by-action count.
            //
            qdr_delivery_decref_CT(core, dlv, "qdr_link_deliver_CT - removed from action");
        }
        return;
    }

    //
    // NOTE: The link->undelivered list does not need to be protected by the
    //       connection's work lock for incoming links.  This protection is only
    //       needed for outgoing links.
    //

    if (DEQ_IS_EMPTY(link->undelivered)) {
        qdr_link_ref_t *temp_rlink = 0;
        qdr_address_t *addr = link->owning_addr;
        if (!addr && dlv->to_addr) {
            qdr_connection_t *conn = link->conn;
            if (conn && conn->tenant_space)
                qd_iterator_annotate_space(dlv->to_addr, conn->tenant_space, conn->tenant_space_len);
            qd_hash_retrieve(core->addr_hash, dlv->to_addr, (void**) &addr);

            if (!addr) {
                //
                // This is an anonymous delivery but the address that it wants sent to is
                // not in this router's address table. We will send this delivery up the
                // anonymous link to the interior router (if this is an edge router).
                // Only edge routers have a non null core->edge_conn_addr
                //
                if (core->edge_conn_addr && link->conn->role != QDR_ROLE_EDGE_CONNECTION) {
                    qdr_address_t *sender_address = core->edge_conn_addr(core->edge_context);
                    if (sender_address) {
                        addr = sender_address;
                    }
                }
            }
            else {
                //
                // (core->edge_conn_addr is non-zero ONLY on edge routers. So there is no need to check if the
                // core->router_mode is edge.
                //
                // The connection on which the delivery arrived should not be QDR_ROLE_EDGE_CONNECTION because
                // we do not want to send it back over the same connections
                //
                if (core->edge_conn_addr && link->conn->role != QDR_ROLE_EDGE_CONNECTION && qdr_is_addr_treatment_multicast(addr)) {
                    qdr_address_t *sender_address = core->edge_conn_addr(core->edge_context);
                    if (sender_address && sender_address != addr) {
                        qdr_link_ref_t *sender_rlink = DEQ_HEAD(sender_address->rlinks);
                        if (sender_rlink) {
                            temp_rlink = new_qdr_link_ref_t();
                            DEQ_ITEM_INIT(temp_rlink);
                            temp_rlink->link = sender_rlink->link;
                            DEQ_INSERT_TAIL(addr->rlinks, temp_rlink);
                        }
                    }
                }
            }
        }

        //
        // Deal with any delivery restrictions for this address.
        //
        if (addr && addr->router_control_only && link->link_type != QD_LINK_CONTROL) {
            qdr_delivery_release_CT(core, dlv);
            qdr_link_issue_credit_CT(core, link, 1, false);
            qdr_delivery_decref_CT(core, dlv, "qdr_link_deliver_CT - removed from action on restricted access");
        } else {
            //
            // Give the action reference to the qdr_link_forward function. Don't decref/incref.
            //
            qdr_link_forward_CT(core, link, dlv, addr, more);
        }

        if (addr && temp_rlink) {
            DEQ_REMOVE(addr->rlinks, temp_rlink);
            free_qdr_link_ref_t(temp_rlink);
        }
    } else {
        //
        // Take the action reference and use it for undelivered.  Don't decref/incref.
        //
        DEQ_INSERT_TAIL(link->undelivered, dlv);
        dlv->where = QDR_DELIVERY_IN_UNDELIVERED;
        qd_log(core->log, QD_LOG_DEBUG, "Delivery transfer:  dlv:%lx qdr_link_deliver_CT: action-list -> undelivered-list", (long) dlv);
    }
}


static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    if (!discard) {
        qdr_in_process_send_to_CT(core,
                                  qdr_field_iterator(action->args.io.address),
                                  action->args.io.message,
                                  action->args.io.exclude_inprocess,
                                  action->args.io.control);
    }

    qdr_field_free(action->args.io.address);
    qd_message_free(action->args.io.message);
}


/**
 * forward an in-process message based on the destination address
 */
void qdr_in_process_send_to_CT(qdr_core_t *core, qd_iterator_t *address, qd_message_t *msg, bool exclude_inprocess, bool control)
{
    qdr_address_t *addr = 0;

    qd_iterator_reset_view(address, ITER_VIEW_ADDRESS_HASH);
    qd_hash_retrieve(core->addr_hash, address, (void**) &addr);
    if (addr) {
        //
        // Forward the message.  We don't care what the fanout count is.
        //
        (void) qdr_forward_message_CT(core, addr, msg, 0, exclude_inprocess, control);
        addr->deliveries_from_container++;
    } else
        qd_log(core->log, QD_LOG_DEBUG, "In-process send to an unknown address");
}


/**
 * Add link-work to provide credit to the link in an IO thread
 */
void qdr_link_issue_credit_CT(qdr_core_t *core, qdr_link_t *link, int credit, bool drain)
{
    assert(link->link_direction == QD_INCOMING);

    bool drain_changed = link->drain_mode ^ drain;
    link->drain_mode   = drain;

    if (link->credit_pending > 0)
        link->credit_pending = link->credit_pending > credit ? link->credit_pending - credit : 0;

    if (!drain_changed && credit == 0)
        return;

    qdr_link_work_drain_action_t drain_action = QDR_LINK_WORK_DRAIN_ACTION_NONE;
    if (drain_changed)
        drain_action = drain ? QDR_LINK_WORK_DRAIN_ACTION_SET : QDR_LINK_WORK_DRAIN_ACTION_CLEAR;

    qdr_connection_t *conn = link->conn;
    sys_mutex_lock(conn->work_lock);
    qdr_link_work_t *work = DEQ_TAIL(link->work_list);
    // can we avoid adding a new work flow item?
    if (work && work->work_type == QDR_LINK_WORK_FLOW
        && (!drain_changed || work->drain_action == drain_action)) {
        work->value += credit;
        sys_mutex_unlock(conn->work_lock);
        qdr_connection_activate_CT(core, conn);

    } else {
        sys_mutex_unlock(conn->work_lock);

        // need a new work flow item
        work = new_qdr_link_work_t();
        ZERO(work);
        work->work_type = QDR_LINK_WORK_FLOW;
        work->value     = credit;
        if (drain_changed)
            work->drain_action = drain_action;
        qdr_link_enqueue_work_CT(core, link, work);
    }
}


/**
 * Attempt to push all of the undelivered deliveries on an incoming link downrange.
 */
void qdr_drain_inbound_undelivered_CT(qdr_core_t *core, qdr_link_t *link, qdr_address_t *addr)
{
    if (DEQ_SIZE(link->undelivered) > 0) {
        //
        // Move all the undelivered to a local list in case not all can be delivered.
        // We don't want to loop here forever putting the same messages on the undelivered
        // list.
        //
        qdr_delivery_list_t deliveries;
        DEQ_MOVE(link->undelivered, deliveries);

        qdr_delivery_t *dlv = DEQ_HEAD(deliveries);
        while (dlv) {
            DEQ_REMOVE_HEAD(deliveries);
            qdr_link_forward_CT(core, link, dlv, addr, false);
            dlv = DEQ_HEAD(deliveries);
        }
    }
}


/**
 * This function should be called after adding a new destination (subscription, local link,
 * or remote node) to an address.  If this address now has exactly one destination (i.e. it
 * transitioned from unreachable to reachable), make sure any unstarted in-links are issued
 * initial credit.
 *
 * Also, check the inlinks to see if there are undelivered messages.  If so, drain them to
 * the forwarder.
 */
void qdr_addr_start_inlinks_CT(qdr_core_t *core, qdr_address_t *addr)
{
    if (qdr_addr_path_count_CT(addr) == 1 || (!!addr->fallback && qdr_addr_path_count_CT(addr->fallback) == 1)) {
        qdr_link_ref_t *ref = DEQ_HEAD(addr->inlinks);
        while (ref) {
            qdr_link_t *link = ref->link;

            //
            // Issue credit to stalled links
            //
            if (link->credit_pending > 0)
                qdr_link_issue_credit_CT(core, link, link->credit_pending, false);

            //
            // Drain undelivered deliveries via the forwarder
            //
            qdr_drain_inbound_undelivered_CT(core, link, addr);

            ref = DEQ_NEXT(ref);
        }

        if (!!addr->fallback_for)
            qdr_addr_start_inlinks_CT(core, addr->fallback_for);
    }
}