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

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);
static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_delete_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard);

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

void qdr_delivery_read_extension_state(qdr_delivery_t *dlv, uint64_t disposition, pn_data_t* disposition_date, bool update_disposition);
void qdr_delivery_copy_extension_state(qdr_delivery_t *src, qdr_delivery_t *dest, bool update_disposition);


//==================================================================================
// 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);
    dlv->link           = link;
    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.connection.delivery = dlv;
    action->args.connection.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);
    dlv->link           = link;
    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.connection.delivery = dlv;
    action->args.connection.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)
{
    if (tag_length > 32)
        return 0;
    
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    dlv->link         = link;
    dlv->msg          = msg;
    dlv->settled      = settled;
    dlv->presettled   = settled;
    dlv->error        = 0;
    dlv->disposition  = 0;

    qdr_delivery_read_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.connection.delivery = dlv;
    action->args.connection.more = !qd_message_receive_complete(msg);
    action->args.connection.tag_length = tag_length;
    memcpy(action->args.connection.tag, tag, tag_length);
    qdr_action_enqueue(link->core, action);
    return dlv;
}


qdr_delivery_t *qdr_deliver_continue(qdr_delivery_t *in_dlv)
{
    qdr_action_t   *action = qdr_action(qdr_deliver_continue_CT, "deliver_continue");
    action->args.connection.delivery = in_dlv;

    qd_message_t *msg = qdr_delivery_message(in_dlv);
    action->args.connection.more = !qd_message_receive_complete(msg);

    // This incref is for the action reference
    qdr_delivery_incref(in_dlv, "qdr_deliver_continue - add to action list");
    qdr_action_enqueue(in_dlv->link->core, action);
    return in_dlv;
}


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) {
        while (credit > 0) {
            sys_mutex_lock(conn->work_lock);
            dlv = DEQ_HEAD(link->undelivered);
            sys_mutex_unlock(conn->work_lock);
            if (dlv) {
                settled = dlv->settled;
                uint64_t new_disp = core->deliver_handler(core->user_context, link, dlv, settled);
                sys_mutex_lock(conn->work_lock);
                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.
                    //
                    DEQ_REMOVE_HEAD(link->undelivered);
                    num_deliveries_completed++;
                    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);
                    }

                    credit--;
                    link->credit_to_core--;<commit_msg>DISPATCH-1112 - Additional fix. Modified qdr_link_process_deliveries() to not continue processing deliveries if detach has been received on that link. This closes #370<commit_after>/*
 * 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 <qpid/dispatch/amqp.h>
#include <stdio.h>
#include <inttypes.h>

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);
static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_delete_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_deliver_continue_CT(qdr_core_t *core, qdr_action_t *action, bool discard);

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

void qdr_delivery_read_extension_state(qdr_delivery_t *dlv, uint64_t disposition, pn_data_t* disposition_date, bool update_disposition);
void qdr_delivery_copy_extension_state(qdr_delivery_t *src, qdr_delivery_t *dest, bool update_disposition);


//==================================================================================
// 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);
    dlv->link           = link;
    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.connection.delivery = dlv;
    action->args.connection.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);
    dlv->link           = link;
    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.connection.delivery = dlv;
    action->args.connection.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)
{
    if (tag_length > 32)
        return 0;
    
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    dlv->link         = link;
    dlv->msg          = msg;
    dlv->settled      = settled;
    dlv->presettled   = settled;
    dlv->error        = 0;
    dlv->disposition  = 0;

    qdr_delivery_read_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.connection.delivery = dlv;
    action->args.connection.more = !qd_message_receive_complete(msg);
    action->args.connection.tag_length = tag_length;
    memcpy(action->args.connection.tag, tag, tag_length);
    qdr_action_enqueue(link->core, action);
    return dlv;
}


qdr_delivery_t *qdr_deliver_continue(qdr_delivery_t *in_dlv)
{
    qdr_action_t   *action = qdr_action(qdr_deliver_continue_CT, "deliver_continue");
    action->args.connection.delivery = in_dlv;

    qd_message_t *msg = qdr_delivery_message(in_dlv);
    action->args.connection.more = !qd_message_receive_complete(msg);

    // This incref is for the action reference
    qdr_delivery_incref(in_dlv, "qdr_deliver_continue - add to action list");
    qdr_action_enqueue(in_dlv->link->core, action);
    return in_dlv;
}


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);
            sys_mutex_unlock(conn->work_lock);
            if (dlv) {
                settled = dlv->settled;
                uint64_t new_disp = core->deliver_handler(core->user_context, link, dlv, settled);
                sys_mutex_lock(conn->work_lock);
                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.
                    //
                    DEQ_REMOVE_HEAD(link->undelivered);
                    num_deliveries_completed++;
                    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);
                    }

                    credit--;
                    link->credit_to_core--;
{'fraction_non_alphanumeric': 0.1265414885046787, 'fraction_numerical': 0.0035942244531201586, 'mean_word_length': 2.824170616113744, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 130, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1331025', 'n_tokens_mistral': 5312, 'n_tokens_neox': 5065, 'n_words': 1120}