/*
 * 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 <stdio.h>
#include <string.h>
#include "dispatch_private.h"
#include <qpid/dispatch/container.h>
#include <qpid/dispatch/server.h>
#include <qpid/dispatch/message.h>
#include <proton/engine.h>
#include <proton/message.h>
#include <qpid/dispatch/ctools.h>
#include <qpid/dispatch/hash.h>
#include <qpid/dispatch/threading.h>
#include <qpid/dispatch/iterator.h>
#include <qpid/dispatch/log.h>
#include <qpid/dispatch/agent.h>
#include "conditionals.h"

struct qd_node_t {
    qd_container_t       *container;
    const qd_node_type_t *ntype;
    char                 *name;
    void                 *context;
    qd_dist_mode_t        supported_dist;
    qd_lifetime_policy_t  life_policy;
};

ALLOC_DECLARE(qd_node_t);
ALLOC_DEFINE(qd_node_t);
ALLOC_DEFINE(qd_link_item_t);


struct qd_link_t {
    pn_link_t *pn_link;
    void      *context;
    qd_node_t *node;
    bool       drain_mode;
};

ALLOC_DECLARE(qd_link_t);
ALLOC_DEFINE(qd_link_t);


struct qd_delivery_t {
    pn_delivery_t *pn_delivery;
    qd_delivery_t *peer;
    void          *context;
    uint64_t       disposition;
    qd_link_t     *link;
    int            in_fifo;
    bool           pending_delete;
};

ALLOC_DECLARE(qd_delivery_t);
ALLOC_DEFINE(qd_delivery_t);


typedef struct qdc_node_type_t {
    DEQ_LINKS(struct qdc_node_type_t);
    const qd_node_type_t *ntype;
} qdc_node_type_t;
DEQ_DECLARE(qdc_node_type_t, qdc_node_type_list_t);

#define QD_CONTAINER_CLASS_CONTAINER  1
#define QD_CONTAINER_CLASS_NODE_TYPE  2
#define QD_CONTAINER_CLASS_NODE       3

typedef struct container_class_t {
    qd_container_t *container;
    int             class_id;
} container_class_t;

struct qd_container_t {
    qd_dispatch_t        *qd;
    qd_log_source_t      *log_source;
    qd_server_t          *server;
    qd_hash_t            *node_type_map;
    qd_hash_t            *node_map;
    sys_mutex_t          *lock;
    qd_node_t            *default_node;
    qdc_node_type_list_t  node_type_list;
    qd_agent_class_t     *class_container;
    qd_agent_class_t     *class_node_type;
    qd_agent_class_t     *class_node;
};

static void setup_outgoing_link(qd_container_t *container, pn_link_t *pn_link)
{
    sys_mutex_lock(container->lock);
    qd_node_t  *node = 0;
    const char *source = pn_terminus_get_address(pn_link_remote_source(pn_link));
    qd_field_iterator_t *iter;
    // TODO - Extract the name from the structured source

    if (source) {
        iter   = qd_field_iterator_string(source, ITER_VIEW_NODE_ID);
        qd_hash_retrieve(container->node_map, iter, (void*) &node);
        qd_field_iterator_free(iter);
    }
    sys_mutex_unlock(container->lock);

    if (node == 0) {
        if (container->default_node)
            node = container->default_node;
        else {
            // Reject the link
            // TODO - When the API allows, add an error message for "no available node"
            pn_link_close(pn_link);
            return;
        }
    }

    qd_link_t *link = new_qd_link_t();
    if (!link) {
        pn_link_close(pn_link);
        return;
    }

    link->pn_link    = pn_link;
    link->context    = 0;
    link->node       = node;
    link->drain_mode = pn_link_get_drain(pn_link);

    pn_link_set_context(pn_link, link);
    node->ntype->outgoing_handler(node->context, link);
}


static void setup_incoming_link(qd_container_t *container, pn_link_t *pn_link)
{
    sys_mutex_lock(container->lock);
    qd_node_t   *node = 0;
    const char  *target = pn_terminus_get_address(pn_link_remote_target(pn_link));
    qd_field_iterator_t *iter;
    // TODO - Extract the name from the structured target

    if (target) {
        iter   = qd_field_iterator_string(target, ITER_VIEW_NODE_ID);
        qd_hash_retrieve(container->node_map, iter, (void*) &node);
        qd_field_iterator_free(iter);
    }
    sys_mutex_unlock(container->lock);

    if (node == 0) {
        if (container->default_node)
            node = container->default_node;
        else {
            // Reject the link
            // TODO - When the API allows, add an error message for "no available node"
            pn_link_close(pn_link);
            return;
        }
    }

    qd_link_t *link = new_qd_link_t();
    if (!link) {
        pn_link_close(pn_link);
        return;
    }

    link->pn_link    = pn_link;
    link->context    = 0;
    link->node       = node;
    link->drain_mode = pn_link_get_drain(pn_link);

    pn_link_set_context(pn_link, link);
    node->ntype->incoming_handler(node->context, link);
}


static int do_writable(pn_link_t *pn_link)
{
    qd_link_t *link = (qd_link_t*) pn_link_get_context(pn_link);
    if (!link)
        return 0;

    qd_node_t *node = link->node;
    if (!node)
        return 0;

    return node->ntype->writable_handler(node->context, link);
}


static void do_receive(pn_delivery_t *pnd)
{
    pn_link_t     *pn_link  = pn_delivery_link(pnd);
    qd_link_t     *link     = (qd_link_t*) pn_link_get_context(pn_link);
    qd_delivery_t *delivery = (qd_delivery_t*) pn_delivery_get_context(pnd);

    if (link) {
        qd_node_t *node = link->node;
        if (node) {
            if (!delivery) {
                delivery = new_qd_delivery_t();
                delivery->pn_delivery    = pnd;
                delivery->peer           = 0;
                delivery->context        = 0;
                delivery->disposition    = 0;
                delivery->link           = link;
                delivery->in_fifo        = 0;
                delivery->pending_delete = false;
                pn_delivery_set_context(pnd, delivery);
            }

            node->ntype->rx_handler(node->context, link, delivery);
            return;
        }
    }

    //
    // Reject the delivery if we couldn't find a node to handle it
    //
    pn_link_advance(pn_link);
    pn_link_flow(pn_link, 1);
    pn_delivery_update(pnd, PN_REJECTED);
    pn_delivery_settle(pnd);
}


static void do_updated(pn_delivery_t *pnd)
{
    pn_link_t     *pn_link  = pn_delivery_link(pnd);
    qd_link_t     *link     = (qd_link_t*) pn_link_get_context(pn_link);
    qd_delivery_t *delivery = (qd_delivery_t*) pn_delivery_get_context(pnd);

    if (link && delivery) {
        qd_node_t *node = link->node;
        if (node)
            node->ntype->disp_handler(node->context, link, delivery);
    }
}


static int close_handler(void* unused, pn_connection_t *conn)
{
    //
    // Close all links, passing False as the 'closed' argument.  These links are not
    // being properly 'detached'.  They are being orphaned.
    //
    pn_link_t *pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE);
    while (pn_link) {
        qd_link_t *link = (qd_link_t*) pn_link_get_context(pn_link);
        qd_node_t *node = link->node;
        if (node && link)
            node->ntype->link_detach_handler(node->context, link, 0);
        pn_link_close(pn_link);
        free_qd_link_t(link);
        pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE);
    }

    // teardown all sessions
    pn_session_t *ssn = pn_session_head(conn, 0);
    while (ssn) {
        pn_session_close(ssn);
        ssn = pn_session_next(ssn, 0);
    }

    // teardown the connection
    pn_connection_close(conn);
    return 0;
}


static int process_handler(qd_container_t *container, void* unused, pn_connection_t *conn)
{
    pn_session_t    *ssn;
    pn_link_t       *pn_link;
    pn_delivery_t   *delivery;
    int              event_count = 0;

    // Step 1: setup the engine's connection, and any sessions and links
    // that may be pending.

    // initialize the connection if it's new
    if (pn_connection_state(conn) & PN_LOCAL_UNINIT) {
        pn_connection_open(conn);
        event_count++;
    }

    // open all pending sessions
    ssn = pn_session_head(conn, PN_LOCAL_UNINIT);
    while (ssn) {
        pn_session_open(ssn);
        ssn = pn_session_next(ssn, PN_LOCAL_UNINIT);
        event_count++;
    }

    // configure and open any pending links
    pn_link = pn_link_head(conn, PN_LOCAL_UNINIT);
    while (pn_link) {
        if (pn_link_is_sender(pn_link))
            setup_outgoing_link(container, pn_link);
        else
            setup_incoming_link(container, pn_link);
        pn_link = pn_link_next(pn_link, PN_LOCAL_UNINIT);
        event_count++;
    }


    // Step 2: Now drain all the pending deliveries from the connection's
    // work queue and process them

    delivery = pn_work_head(conn);
    while (delivery) {
        if (pn_delivery_readable(delivery))
            do_receive(delivery);

        if (pn_delivery_updated(delivery)) {
            do_updated(delivery);
            pn_delivery_clear(delivery);
        }
        delivery = pn_work_next(delivery);
        event_count++;
    }

    //
    // Step 2.5: Call the attached node's writable handler for all active links
    // on the connection.  Note that in Dispatch, links are considered
    // bidirectional.  Incoming and outgoing only pertains to deliveries and
    // deliveries are a subset of the traffic that flows both directions on links.
    //
    pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE);
    while (pn_link) {
        assert(pn_session_connection(pn_link_session(pn_link)) == conn);
        event_count += do_writable(pn_link);
        pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE);
    }

    // Step 3: Clean up any links or sessions that have been closed by the
    // remote.  If the connection has been closed remotely, clean that up
    // also.

    // teardown any terminating links
    pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
    while (pn_link) {
        qd_link_t *link = (qd_link_t*) pn_link_get_context(pn_link);
        qd_node_t *node = link->node;
        if (node)
            node->ntype->link_detach_handler(node->context, link, 1); // TODO - get 'closed' from detach message
        pn_link_close(pn_link);
        pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
        event_count++;
    }

    // teardown any terminating sessions
    ssn = pn_session_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
    while (ssn) {
        pn_session_close(ssn);
        ssn = pn_session_next(ssn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED);
        event_count++;
    }

    // teardown the connection if it's terminating
    if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) {
        pn_connection_close(conn);
        event_count++;
    }

    return event_count;
}


static void open_handler(qd_container_t *container, qd_connection_t *conn, qd_direction_t dir)
{
    const qd_node_type_t *nt;

    //
    // Note the locking structure in this function.  Generally this would be unsafe, but since
    // this particular list is only ever appended to and never has items inserted or deleted,
    // this usage is safe in this case.
    //
    sys_mutex_lock(container->lock);
    qdc_node_type_t *nt_item = DEQ_HEAD(container->node_type_list);
    sys_mutex_unlock(container->lock);

    pn_connection_open(qd_connection_pn(conn));

    while (nt_item) {
        nt = nt_item->ntype;
        if (dir == QD_INCOMING) {
            if (nt->inbound_conn_open_handler)
                nt->inbound_conn_open_handler(nt->type_context, conn);
        } else {
            if (nt->outbound_conn_open_handler)
                nt->outbound_conn_open_handler(nt->type_context, conn);
        }

        sys_mutex_lock(container->lock);
        nt_item = DEQ_NEXT(nt_item);
        sys_mutex_unlock(container->lock);
    }
}


static int handler(void *handler_context, void *conn_context, qd_conn_event_t event, qd_connection_t *qd_conn)
{
    qd_container_t  *container = (qd_container_t*) handler_context;
    pn_connection_t *conn      = qd_connection_pn(qd_conn);

    switch (event) {
    case QD_CONN_EVENT_LISTENER_OPEN:  open_handler(container, qd_conn, QD_INCOMING);   break;
    case QD_CONN_EVENT_CONNECTOR_OPEN: open_handler(container, qd_conn, QD_OUTGOING);   break;
    case QD_CONN_EVENT_CLOSE:          return close_handler(conn_context, conn);
    case QD_CONN_EVENT_PROCESS:        return process_handler(container, conn_context, conn);
    }

    return 0;
}


static void container_schema_handler(void *context, void *correlator)
{
    container_class_t *cls = (container_class_t*) context;
    switch (cls->class_id) {
    case QD_CONTAINER_CLASS_CONTAINER:
        qd_agent_value_string(correlator, 0, "node_type_count");
        qd_agent_value_string(correlator, 0, "node_count");
        qd_agent_value_string(correlator, 0, "default_node_type");
        break;

    case QD_CONTAINER_CLASS_NODE_TYPE:
    case QD_CONTAINER_CLASS_NODE:
        break;
    }
}


static void container_query_handler(void* context, const char *id, void *correlator)
{
    container_class_t *cls = (container_class_t*) context;

    if (cls->class_id == QD_CONTAINER_CLASS_CONTAINER) {
        qd_agent_value_uint(correlator, "node_type_count", qd_hash_size(cls->container->node_type_map));
        qd_agent_value_uint(correlator, "node_count",      qd_hash_size(cls->container->node_map));
        if (cls->container->default_node)
            qd_agent_value_string(correlator, "default_node_type", cls->container->default_node->ntype->type_name);
        else
            qd_agent_value_null(correlator, "default_node_type");
        qd_agent_value_complete(correlator, false);

    } else if (cls->class_id == QD_CONTAINER_CLASS_NODE_TYPE) {

    } else if (cls->class_id == QD_CONTAINER_CLASS_NODE) {

    }
}


qd_agent_class_t *setup_class(qd_container_t *container, const char *fqname, int id)
{
    container_class_t *cls = NEW(container_class_t);
    cls->container = container;
    cls->class_id  = id;

    return qd_agent_register_class(container->qd, fqname, cls,
                                   container_schema_handler,
                                   container_query_handler);
}


qd_container_t *qd_container(qd_dispatch_t *qd)
{
    qd_container_t *container = NEW(qd_container_t);

    container->qd            = qd;
    container->log_source    = qd_log_source("CONTAINER");
    container->server        = qd->server;
    container->node_type_map = qd_hash(6,  4, 1);  // 64 buckets, item batches of 4
    container->node_map      = qd_hash(10, 32, 0); // 1K buckets, item batches of 32
    container->lock          = sys_mutex();
    container->default_node  = 0;
    DEQ_INIT(container->node_type_list);

    qd_log(container->log_source, QD_LOG_TRACE, "Container Initializing");
    qd_server_set_conn_handler(qd, handler, container);

    return container;
}


void qd_container_setup_agent(qd_dispatch_t *qd)
{
    qd->container->class_container =
        setup_class(qd->container, "org.apache.qpid.dispatch.container", QD_CONTAINER_CLASS_CONTAINER);
    qd->container->class_node_type =
        setup_class(qd->container, "org.apache.qpid.dispatch.container.node_type", QD_CONTAINER_CLASS_NODE_TYPE);
    qd->container->class_node =
        setup_class(qd->container, "org.apache.qpid.dispatch.container.node", QD_CONTAINER_CLASS_NODE);
}


void qd_container_free(qd_container_t *container)
{
    // TODO - Free the nodes
    // TODO - Free the node types
    sys_mutex_free(container->lock);
    free(container);
}


int qd_container_register_node_type(qd_dispatch_t *qd, const qd_node_type_t *nt)
{
    qd_container_t *container = qd->container;

    int result;
    qd_field_iterator_t *iter = qd_field_iterator_string(nt->type_name, ITER_VIEW_ALL);
    qdc_node_type_t     *nt_item = NEW(qdc_node_type_t);
    DEQ_ITEM_INIT(nt_item);
    nt_item->ntype = nt;

    sys_mutex_lock(container->lock);
    result = qd_hash_insert_const(container->node_type_map, iter, nt, 0);
    DEQ_INSERT_TAIL(container->node_type_list, nt_item);
    sys_mutex_unlock(container->lock);

    qd_field_iterator_free(iter);
    if (result < 0)
        return result;
    qd_log(container->log_source, QD_LOG_TRACE, "Node Type Registered - %s", nt->type_name);

    return 0;
}


qd_node_t *qd_container_set_default_node_type(qd_dispatch_t        *qd,
                                              const qd_node_type_t *nt,
                                              void                 *context,
                                              qd_dist_mode_t        supported_dist)
{
    qd_container_t *container = qd->container;

    if (container->default_node)
        qd_container_destroy_node(container->default_node);

    if (nt) {
        container->default_node = qd_container_create_node(qd, nt, 0, context, supported_dist, QD_LIFE_PERMANENT);
        qd_log(container->log_source, QD_LOG_TRACE, "Node of type '%s' installed as default node", nt->type_name);
    } else {
        container->default_node = 0;
        qd_log(container->log_source, QD_LOG_TRACE, "Default node removed");
    }

    return container->default_node;
}


qd_node_t *qd_container_create_node(qd_dispatch_t        *qd,
                                    const qd_node_type_t *nt,
                                    const char           *name,
                                    void                 *context,
                                    qd_dist_mode_t        supported_dist,
                                    qd_lifetime_policy_t  life_policy)
{
    qd_container_t *container = qd->container;
    int result;
    qd_node_t *node = new_qd_node_t();
    if (!node)
        return 0;

    node->container      = container;
    node->ntype          = nt;
    node->name           = 0;
    node->context        = context;
    node->supported_dist = supported_dist;
    node->life_policy    = life_policy;

    if (name) {
        qd_field_iterator_t *iter = qd_field_iterator_string(name, ITER_VIEW_ALL);
        sys_mutex_lock(container->lock);
        result = qd_hash_insert(container->node_map, iter, node, 0);
        sys_mutex_unlock(container->lock);
        qd_field_iterator_free(iter);
        if (result < 0) {
            free_qd_node_t(node);
            return 0;
        }

        node->name = (char*) malloc(strlen(name) + 1);
        strcpy(node->name, name);
    }

    if (name)
        qd_log(container->log_source, QD_LOG_TRACE, "Node of type '%s' created with name '%s'", nt->type_name, name);

    return node;
}


void qd_container_destroy_node(qd_node_t *node)
{
    qd_container_t *container = node->container;

    if (node->name) {
        qd_field_iterator_t *iter = qd_field_iterator_string(node->name, ITER_VIEW_ALL);
        sys_mutex_lock(container->lock);
        qd_hash_remove(container->node_map, iter);
        sys_mutex_unlock(container->lock);
        qd_field_iterator_free(iter);
        free(node->name);
    }

    free_qd_node_t(node);
}


void qd_container_node_set_context(qd_node_t *node, void *node_context)
{
    node->context = node_context;
}


qd_dist_mode_t qd_container_node_get_dist_modes(const qd_node_t *node)
{
    return node->supported_dist;
}


qd_lifetime_policy_t qd_container_node_get_life_policy(const qd_node_t *node)
{
    return node->life_policy;
}


qd_link_t *qd_link(qd_node_t *node, qd_connection_t *conn, qd_direction_t dir, const char* name)
{
    pn_session_t *sess = pn_session(qd_connection_pn(conn));
    qd_link_t    *link = new_qd_link_t();

    if (dir == QD_OUTGOING)
        link->pn_link = pn_sender(sess, name);
    else
        link->pn_link = pn_receiver(sess, name);

    link->context    = node->context;
    link->node       = node;
    link->drain_mode = pn_link_get_drain(link->pn_link);

    pn_link_set_context(link->pn_link, link);

    pn_session_open(sess);

    return link;
}


void qd_link_set_context(qd_link_t *link, void *context)
{
    link->context = context;
}


void *qd_link_get_context(qd_link_t *link)
{
    return link->context;
}


void qd_link_set_conn_context(qd_link_t *link, void *context)
{
    pn_session_t *pn_sess = pn_link_session(link->pn_link);
    if (!pn_sess)
        return;
    pn_connection_t *pn_conn = pn_session_connection(pn_sess);
    if (!pn_conn)
        return;
    qd_connection_t *conn = (qd_connection_t*) pn_connection_get_context(pn_conn);
    if (!conn)
        return;
    qd_connection_set_link_context(conn, context);
}


void *qd_link_get_conn_context(qd_link_t *link)
{
    pn_session_t *pn_sess = pn_link_session(link->pn_link);
    if (!pn_sess)
        return 0;
    pn_connection_t *pn_conn = pn_session_connection(pn_sess);
    if (!pn_conn)
        return 0;
    qd_connection_t *conn = (qd_connection_t*) pn_connection_get_context(pn_conn);
    if (!conn)
        return 0;
    return qd_connection_get_link_context(conn);
}


pn_link_t *qd_link_pn(qd_link_t *link)
{
    return link->pn_link;
}


qd_connection_t *qd_link_connection(qd_link_t *link)
{
    if (!link || !link->pn_link)
        return 0;

    pn_session_t *sess = pn_link_session(link->pn_link);
    if (!sess)
        return 0;

    pn_connection_t *conn = pn_session_connection(sess);
    if (!conn)
        return 0;

    qd_connection_t *ctx = pn_connection_get_context(conn);
    if (!ctx)
        return 0;

    return ctx;
}


pn_terminus_t *qd_link_source(qd_link_t *link)
{
    return pn_link_source(link->pn_link);
}


pn_terminus_t *qd_link_target(qd_link_t *link)
{
    return pn_link_target(link->pn_link);
}


pn_terminus_t *qd_link_remote_source(qd_link_t *link)
{
    return pn_link_remote_source(link->pn_link);
}


pn_terminus_t *qd_link_remote_target(qd_link_t *link)
{
    return pn_link_remote_target(link->pn_link);
}


void qd_link_activate(qd_link_t *link)
{
    if (!link || !link->pn_link || pn_link_state(link->pn_link) != (PN_LOCAL_ACTIVE|PN_REMOTE_ACTIVE))
        return;

    pn_session_t *sess = pn_link_session(link->pn_link);
    if (!sess || pn_session_state(sess) != (PN_LOCAL_ACTIVE|PN_REMOTE_ACTIVE))
        return;

    pn_connection_t *conn = pn_session_connection(sess);
    if (!conn || pn_connection_state(conn) != (PN_LOCAL_ACTIVE|PN_REMOTE_ACTIVE))
        return;

    qd_connection_t *ctx = pn_connection_get_context(conn);
    if (!ctx)
        return;

    qd_server_activate(ctx);
}


void qd_link_close(qd_link_t *link)
{
    pn_link_close(link->pn_link);
}


bool qd_link_drain_changed(qd_link_t *link, bool *mode)
{
    bool pn_mode = pn_link_get_drain(link->pn_link);
    bool changed = pn_mode != link->drain_mode;

    *mode = pn_mode;
    if (changed)
        link->drain_mode = pn_mode;
    return changed;
}


qd_delivery_t *qd_delivery(qd_link_t *link, pn_delivery_tag_t tag)
{
    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;

    qd_delivery_t *delivery = new_qd_delivery_t();
    delivery->pn_delivery    = pnd;
    delivery->peer           = 0;
    delivery->context        = 0;
    delivery->disposition    = 0;
    delivery->link           = link;
    delivery->in_fifo        = 0;
    delivery->pending_delete = false;
    pn_delivery_set_context(pnd, delivery);

    return delivery;
}


void qd_delivery_free_LH(qd_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;
    }

    assert(!delivery->peer);

    if (delivery->in_fifo)
        delivery->pending_delete = true;
    else {
        free_qd_delivery_t(delivery);
    }
}


void qd_delivery_link_peers_LH(qd_delivery_t *right, qd_delivery_t *left)
{
    right->peer = left;
    left->peer  = right;
}


void qd_delivery_unlink_LH(qd_delivery_t *delivery)
{
    if (delivery->peer) {
        delivery->peer->peer = 0;
        delivery->peer       = 0;
    }
}


void qd_delivery_fifo_enter_LH(qd_delivery_t *delivery)
{
    delivery->in_fifo++;
}


bool qd_delivery_fifo_exit_LH(qd_delivery_t *delivery)
{
    delivery->in_fifo--;
    if (delivery->in_fifo == 0 && delivery->pending_delete) {
        free_qd_delivery_t(delivery);
        return false;
    }

    return true;
}


void qd_delivery_set_context(qd_delivery_t *delivery, void *context)
{
    delivery->context = context;
}


void *qd_delivery_context(qd_delivery_t *delivery)
{
    return delivery->context;
}


qd_delivery_t *qd_delivery_peer(qd_delivery_t *delivery)
{
    return delivery->peer;
}


pn_delivery_t *qd_delivery_pn(qd_delivery_t *delivery)
{
    return delivery->pn_delivery;
}


void qd_delivery_settle(qd_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_delivery_settled(qd_delivery_t *delivery)
{
    return pn_delivery_settled(delivery->pn_delivery);
}


bool qd_delivery_disp_changed(qd_delivery_t *delivery)
{
    return delivery->disposition != pn_delivery_remote_state(delivery->pn_delivery);
}


uint64_t qd_delivery_disp(qd_delivery_t *delivery)
{
    delivery->disposition = pn_delivery_remote_state(delivery->pn_delivery);
    return delivery->disposition;
}


qd_link_t *qd_delivery_link(qd_delivery_t *delivery)
{
    return delivery->link;
}