| /* | |
| * Licensed to the Apache Software Foundation (ASF) under one | |
| * or more contributor license agreements. See the NOTICE file | |
| * distributed with this work for additional information | |
| * regarding copyright ownership. The ASF licenses this file | |
| * to you under the Apache License, Version 2.0 (the | |
| * "License"); you may not use this file except in compliance | |
| * with the License. You may obtain a copy of the License at | |
| * | |
| * http://www.apache.org/licenses/LICENSE-2.0 | |
| * | |
| * Unless required by applicable law or agreed to in writing, | |
| * software distributed under the License is distributed on an | |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| * KIND, either express or implied. See the License for the | |
| * specific language governing permissions and limitations | |
| * under the License. | |
| */ | |
| #include "qpid/dispatch/container.h" | |
| #include "dispatch_private.h" | |
| #include "policy.h" | |
| #include "qpid/dispatch/amqp.h" | |
| #include "qpid/dispatch/ctools.h" | |
| #include "qpid/dispatch/hash.h" | |
| #include "qpid/dispatch/iterator.h" | |
| #include "qpid/dispatch/log.h" | |
| #include "qpid/dispatch/message.h" | |
| #include "qpid/dispatch/server.h" | |
| #include "qpid/dispatch/threading.h" | |
| #include <proton/connection.h> | |
| #include <proton/engine.h> | |
| #include <proton/event.h> | |
| #include <proton/message.h> | |
| #include <inttypes.h> | |
| #include <stdio.h> | |
| #include <string.h> | |
| /** Instance of a node type in a container */ | |
| struct qd_node_t { | |
| DEQ_LINKS(qd_node_t); | |
| qd_container_t *container; | |
| const qd_node_type_t *ntype; ///< Type of node, defines callbacks. | |
| char *name; | |
| void *context; | |
| qd_dist_mode_t supported_dist; | |
| qd_lifetime_policy_t life_policy; | |
| }; | |
| DEQ_DECLARE(qd_node_t, qd_node_list_t); | |
| ALLOC_DECLARE(qd_node_t); | |
| ALLOC_DEFINE(qd_node_t); | |
| /** Encapsulates a proton link for sending and receiving messages */ | |
| struct qd_link_t { | |
| DEQ_LINKS(qd_link_t); | |
| pn_session_t *pn_sess; | |
| pn_link_t *pn_link; | |
| qd_direction_t direction; | |
| void *context; | |
| qd_node_t *node; | |
| qd_alloc_safe_ptr_t incoming_msg; // DISPATCH-1690: for cleanup | |
| pn_snd_settle_mode_t remote_snd_settle_mode; | |
| qd_link_ref_list_t ref_list; | |
| bool q2_limit_unbounded; | |
| bool q3_blocked; | |
| DEQ_LINKS_N(Q3, qd_link_t); ///< Q3 blocked links | |
| uint64_t link_id; | |
| }; | |
| ALLOC_DEFINE(qd_link_t); | |
| ALLOC_DEFINE(qd_link_ref_t); | |
| /** Encapsulates a proton session */ | |
| struct qd_session_t { | |
| DEQ_LINKS(qd_session_t); | |
| pn_session_t *pn_sess; | |
| qd_link_list_t q3_blocked_links; ///< Q3 blocked if !empty | |
| }; | |
| // macros to work around PROTON-2184 (see container.h) | |
| // | |
| static inline void qd_session_link_pn(qd_session_t *qd_ssn, pn_session_t *pn_ssn) | |
| { | |
| assert(qd_ssn && pn_ssn); | |
| pn_session_set_context(pn_ssn, qd_ssn); | |
| } | |
| static inline void qd_session_unlink_pn(qd_session_t *qd_ssn, pn_session_t *pn_ssn) | |
| { | |
| assert(qd_ssn); | |
| pn_session_set_context(pn_ssn, 0); | |
| } | |
| ALLOC_DECLARE(qd_session_t); | |
| ALLOC_DEFINE(qd_session_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); | |
| 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; | |
| qd_node_list_t nodes; | |
| sys_mutex_t *lock; | |
| qd_node_t *default_node; | |
| qdc_node_type_list_t node_type_list; | |
| qd_link_list_t links; | |
| }; | |
| ALLOC_DEFINE(qd_pn_free_link_session_t); | |
| static void setup_outgoing_link(qd_container_t *container, pn_link_t *pn_link) | |
| { | |
| qd_node_t *node = container->default_node; | |
| if (node == 0) { | |
| pn_condition_t *cond = pn_link_condition(pn_link); | |
| pn_condition_set_name(cond, QD_AMQP_COND_NOT_FOUND); | |
| pn_condition_set_description(cond, "Source node does not exist"); | |
| pn_link_close(pn_link); | |
| return; | |
| } | |
| qd_link_t *link = new_qd_link_t(); | |
| if (!link) { | |
| pn_condition_t *cond = pn_link_condition(pn_link); | |
| pn_condition_set_name(cond, QD_AMQP_COND_INTERNAL_ERROR); | |
| pn_condition_set_description(cond, "Insufficient memory"); | |
| pn_link_close(pn_link); | |
| return; | |
| } | |
| ZERO(link); | |
| sys_mutex_lock(container->lock); | |
| DEQ_INSERT_TAIL(container->links, link); | |
| sys_mutex_unlock(container->lock); | |
| link->pn_sess = pn_link_session(pn_link); | |
| link->pn_link = pn_link; | |
| link->direction = QD_OUTGOING; | |
| link->context = 0; | |
| link->node = node; | |
| link->remote_snd_settle_mode = pn_link_remote_snd_settle_mode(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, uint64_t max_size) | |
| { | |
| qd_node_t *node = container->default_node; | |
| if (node == 0) { | |
| pn_condition_t *cond = pn_link_condition(pn_link); | |
| pn_condition_set_name(cond, QD_AMQP_COND_NOT_FOUND); | |
| pn_condition_set_description(cond, "Target node does not exist"); | |
| pn_link_close(pn_link); | |
| return; | |
| } | |
| qd_link_t *link = new_qd_link_t(); | |
| if (!link) { | |
| pn_condition_t *cond = pn_link_condition(pn_link); | |
| pn_condition_set_name(cond, QD_AMQP_COND_INTERNAL_ERROR); | |
| pn_condition_set_description(cond, "Insufficient memory"); | |
| pn_link_close(pn_link); | |
| return; | |
| } | |
| ZERO(link); | |
| sys_mutex_lock(container->lock); | |
| DEQ_INSERT_TAIL(container->links, link); | |
| sys_mutex_unlock(container->lock); | |
| link->pn_sess = pn_link_session(pn_link); | |
| link->pn_link = pn_link; | |
| link->direction = QD_INCOMING; | |
| link->context = 0; | |
| link->node = node; | |
| link->remote_snd_settle_mode = pn_link_remote_snd_settle_mode(pn_link); | |
| if (max_size) { | |
| pn_link_set_max_message_size(pn_link, max_size); | |
| } | |
| pn_link_set_context(pn_link, link); | |
| node->ntype->incoming_handler(node->context, link); | |
| } | |
| static void handle_link_open(qd_container_t *container, pn_link_t *pn_link) | |
| { | |
| qd_link_t *link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (link == 0) | |
| return; | |
| if (link->node->ntype->link_attach_handler) | |
| link->node->ntype->link_attach_handler(link->node->context, link); | |
| } | |
| static void do_receive(pn_link_t *pn_link, pn_delivery_t *pnd) | |
| { | |
| qd_link_t *link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (link) { | |
| qd_node_t *node = link->node; | |
| if (node) { | |
| while (true) { | |
| if (!node->ntype->rx_handler(node->context, link)) | |
| break; | |
| } | |
| 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); | |
| if (link) { | |
| qd_node_t *node = link->node; | |
| if (node) | |
| node->ntype->disp_handler(node->context, link, pnd); | |
| } | |
| } | |
| static void notify_opened(qd_container_t *container, qd_connection_t *conn, void *context) | |
| { | |
| 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); | |
| while (nt_item) { | |
| nt = nt_item->ntype; | |
| if (qd_connection_inbound(conn)) { | |
| if (nt->inbound_conn_opened_handler) | |
| nt->inbound_conn_opened_handler(nt->type_context, conn, context); | |
| } else { | |
| if (nt->outbound_conn_opened_handler) | |
| nt->outbound_conn_opened_handler(nt->type_context, conn, context); | |
| } | |
| sys_mutex_lock(container->lock); | |
| nt_item = DEQ_NEXT(nt_item); | |
| sys_mutex_unlock(container->lock); | |
| } | |
| } | |
| void policy_notify_opened(void *container, qd_connection_t *conn, void *context) | |
| { | |
| notify_opened((qd_container_t *)container, (qd_connection_t *)conn, context); | |
| } | |
| static void notify_closed(qd_container_t *container, qd_connection_t *conn, void *context) | |
| { | |
| 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. | |
| // | |
| // This assumes that pointer assignment is atomic, which it is on most platforms. | |
| // | |
| sys_mutex_lock(container->lock); | |
| qdc_node_type_t *nt_item = DEQ_HEAD(container->node_type_list); | |
| sys_mutex_unlock(container->lock); | |
| while (nt_item) { | |
| nt = nt_item->ntype; | |
| if (nt->conn_closed_handler) | |
| nt->conn_closed_handler(nt->type_context, conn, context); | |
| sys_mutex_lock(container->lock); | |
| nt_item = DEQ_NEXT(nt_item); | |
| sys_mutex_unlock(container->lock); | |
| } | |
| } | |
| static void close_links(qd_container_t *container, pn_connection_t *conn, bool print_log) | |
| { | |
| pn_link_t *pn_link = pn_link_head(conn, 0); | |
| while (pn_link) { | |
| qd_link_t *qd_link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (qd_link && qd_link_get_context(qd_link) == 0) { | |
| pn_link_set_context(pn_link, 0); | |
| pn_link = pn_link_next(pn_link, 0); | |
| qd_link_free(qd_link); | |
| continue; | |
| } | |
| if (qd_link && qd_link->node) { | |
| qd_node_t *node = qd_link->node; | |
| if (print_log) | |
| qd_log(container->log_source, QD_LOG_DEBUG, | |
| "Aborting link '%s' due to parent connection end", | |
| pn_link_name(pn_link)); | |
| node->ntype->link_detach_handler(node->context, qd_link, QD_LOST); | |
| } | |
| pn_link = pn_link_next(pn_link, 0); | |
| } | |
| } | |
| /** cleanup_link | |
| * Prior to freeing qd_link_t - safe to call during shutdown | |
| */ | |
| static void cleanup_link(qd_link_t *link) | |
| { | |
| if (link) { | |
| if (link->q3_blocked) | |
| qd_link_q3_unblock(link); | |
| if (link->pn_link) { | |
| pn_link_set_context(link->pn_link, NULL); | |
| link->pn_link = 0; | |
| } | |
| link->pn_sess = 0; | |
| // cleanup any inbound message that has not been forwarded | |
| qd_message_t *msg = qd_alloc_deref_safe_ptr(&link->incoming_msg); | |
| if (msg) { | |
| qd_message_free(msg); | |
| } | |
| } | |
| } | |
| static int close_handler(qd_container_t *container, pn_connection_t *conn, qd_connection_t* qd_conn) | |
| { | |
| // | |
| // Close all links, passing QD_LOST as the reason. These links are not | |
| // being properly 'detached'. They are being orphaned. | |
| // | |
| if (qd_conn) | |
| qd_conn->closed = true; | |
| close_links(container, conn, true); | |
| qd_session_cleanup(qd_conn); | |
| if (qd_conn) | |
| notify_closed(container, qd_conn, qd_connection_get_context(qd_conn)); | |
| return 0; | |
| } | |
| static void writable_handler(qd_container_t *container, pn_connection_t *conn, qd_connection_t* qd_conn) | |
| { | |
| 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); | |
| while (nt_item) { | |
| nt = nt_item->ntype; | |
| if (nt->writable_handler) | |
| nt->writable_handler(nt->type_context, qd_conn, 0); | |
| sys_mutex_lock(container->lock); | |
| nt_item = DEQ_NEXT(nt_item); | |
| sys_mutex_unlock(container->lock); | |
| } | |
| } | |
| /** | |
| * Returns true if the free_link already exists in free_link_list, false otherwise | |
| */ | |
| static bool link_exists(qd_pn_free_link_session_list_t *free_list, pn_link_t *free_link) | |
| { | |
| qd_pn_free_link_session_t *free_item = DEQ_HEAD(*free_list); | |
| while(free_item) { | |
| if (free_item->pn_link == free_link) | |
| return true; | |
| free_item = DEQ_NEXT(free_item); | |
| } | |
| return false; | |
| } | |
| /** | |
| * Returns true if the free_session already exists in free_session_list, false otherwise | |
| */ | |
| static bool session_exists(qd_pn_free_link_session_list_t *free_list, pn_session_t *free_session) | |
| { | |
| qd_pn_free_link_session_t *free_item = DEQ_HEAD(*free_list); | |
| while(free_item) { | |
| if (free_item->pn_session == free_session) | |
| return true; | |
| free_item = DEQ_NEXT(free_item); | |
| } | |
| return false; | |
| } | |
| static void add_session_to_free_list(qd_pn_free_link_session_list_t *free_link_session_list, pn_session_t *ssn) | |
| { | |
| if (!session_exists(free_link_session_list, ssn)) { | |
| qd_pn_free_link_session_t *to_free = new_qd_pn_free_link_session_t(); | |
| DEQ_ITEM_INIT(to_free); | |
| to_free->pn_session = ssn; | |
| to_free->pn_link = 0; | |
| DEQ_INSERT_TAIL(*free_link_session_list, to_free); | |
| } | |
| } | |
| static void add_link_to_free_list(qd_pn_free_link_session_list_t *free_link_session_list, pn_link_t *pn_link) | |
| { | |
| if (!link_exists(free_link_session_list, pn_link)) { | |
| qd_pn_free_link_session_t *to_free = new_qd_pn_free_link_session_t(); | |
| DEQ_ITEM_INIT(to_free); | |
| to_free->pn_link = pn_link; | |
| to_free->pn_session = 0; | |
| DEQ_INSERT_TAIL(*free_link_session_list, to_free); | |
| } | |
| } | |
| /* | |
| * The need for these lists may indicate a router bug, where the router is | |
| * using links/sessions after they are freed. Investigate and simplify if | |
| * possible. | |
| */ | |
| void qd_conn_event_batch_complete(qd_container_t *container, qd_connection_t *qd_conn, bool conn_closed) | |
| { | |
| qd_pn_free_link_session_t *to_free = DEQ_HEAD(qd_conn->free_link_session_list); | |
| while(to_free) { | |
| if (!conn_closed) { | |
| if (to_free->pn_link) { | |
| qd_link_t *qd_link = (qd_link_t*) pn_link_get_context(to_free->pn_link); | |
| if (qd_link) { | |
| qd_link->pn_link = 0; | |
| } | |
| pn_link_set_context(to_free->pn_link, 0); | |
| pn_link_free(to_free->pn_link); | |
| } | |
| if (to_free->pn_session) | |
| pn_session_free(to_free->pn_session); | |
| } | |
| DEQ_REMOVE_HEAD(qd_conn->free_link_session_list); | |
| free_qd_pn_free_link_session_t(to_free); | |
| to_free = DEQ_HEAD(qd_conn->free_link_session_list); | |
| } | |
| if (!conn_closed) { | |
| writable_handler(container, qd_conn->pn_conn, qd_conn); | |
| } | |
| } | |
| void qd_container_handle_event(qd_container_t *container, pn_event_t *event, | |
| pn_connection_t *conn, qd_connection_t *qd_conn) | |
| { | |
| pn_session_t *ssn = NULL; | |
| pn_link_t *pn_link = NULL; | |
| qd_link_t *qd_link = NULL; | |
| pn_delivery_t *delivery = NULL; | |
| switch (pn_event_type(event)) { | |
| case PN_CONNECTION_REMOTE_OPEN : | |
| qd_connection_set_user(qd_conn); | |
| qd_conn->open_container = (void *)container; | |
| if (pn_connection_state(conn) & PN_LOCAL_UNINIT) { | |
| // This Open is an externally initiated connection | |
| // Let policy engine decide | |
| /* TODO aconway 2017-04-11: presently the policy test is run | |
| * in the current thread. | |
| * | |
| * If/when the policy test can run in another thread, the connection | |
| * can be stalled by saving the current pn_event_batch and passing it | |
| * to pn_proactor_done() when the policy check is complete. Note we | |
| * can't run the policy check as a deferred function on the current | |
| * connection since by stalling the current connection it will never be | |
| * run, so we need some other thread context to run it in. | |
| */ | |
| qd_policy_amqp_open(qd_conn); | |
| } else { | |
| // This Open is in response to an internally initiated connection | |
| qd_policy_amqp_open_connector(qd_conn); | |
| } | |
| break; | |
| case PN_CONNECTION_REMOTE_CLOSE : | |
| if (qd_conn) | |
| qd_conn->closed = true; | |
| if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) { | |
| close_links(container, conn, false); | |
| qd_session_cleanup(qd_conn); | |
| pn_connection_close(conn); | |
| qd_conn_event_batch_complete(container, qd_conn, true); | |
| } else if (pn_connection_state(conn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) { | |
| close_links(container, conn, false); | |
| qd_session_cleanup(qd_conn); | |
| notify_closed(container, qd_conn, qd_connection_get_context(qd_conn)); | |
| qd_conn_event_batch_complete(container, qd_conn, true); | |
| } | |
| break; | |
| case PN_SESSION_REMOTE_OPEN : | |
| if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) { | |
| ssn = pn_event_session(event); | |
| if (pn_session_state(ssn) & PN_LOCAL_UNINIT) { | |
| // remote created new session | |
| assert(qd_session_from_pn(ssn) == 0); | |
| qd_session_t *qd_ssn = qd_session(ssn); | |
| if (!qd_ssn) { | |
| pn_condition_t *cond = pn_session_condition(ssn); | |
| pn_condition_set_name(cond, QD_AMQP_COND_INTERNAL_ERROR); | |
| pn_condition_set_description(cond, "Insufficient memory"); | |
| pn_session_close(ssn); | |
| break; | |
| } | |
| if (qd_conn->policy_settings) { | |
| if (!qd_policy_approve_amqp_session(ssn, qd_conn)) { | |
| qd_session_free(qd_ssn); | |
| break; | |
| } | |
| qd_conn->n_sessions++; | |
| } | |
| qd_session_link_pn(qd_ssn, ssn); | |
| qd_policy_apply_session_settings(ssn, qd_conn); | |
| pn_session_open(ssn); | |
| } | |
| } | |
| break; | |
| case PN_SESSION_LOCAL_CLOSE : | |
| ssn = pn_event_session(event); | |
| for (int i = 0; i < QD_SSN_CLASS_COUNT; ++i) { | |
| if (ssn == qd_conn->pn_sessions[i]) { | |
| qd_conn->pn_sessions[i] = 0; | |
| break; | |
| } | |
| } | |
| pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); | |
| while (pn_link) { | |
| if (pn_link_session(pn_link) == ssn) { | |
| qd_link_t *qd_link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (qd_link) | |
| qd_link->pn_link = 0; | |
| } | |
| pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); | |
| } | |
| if (pn_session_state(ssn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(ssn); | |
| qd_session_free(qd_ssn); | |
| add_session_to_free_list(&qd_conn->free_link_session_list, ssn); | |
| } | |
| break; | |
| case PN_SESSION_REMOTE_CLOSE : | |
| ssn = pn_event_session(event); | |
| for (int i = 0; i < QD_SSN_CLASS_COUNT; ++i) { | |
| if (ssn == qd_conn->pn_sessions[i]) { | |
| qd_conn->pn_sessions[i] = 0; | |
| break; | |
| } | |
| } | |
| if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) { | |
| if (pn_session_state(ssn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) { | |
| // remote has nuked our session. Check for any links that were | |
| // left open and forcibly detach them, since no detaches will | |
| // arrive on this session. | |
| pn_connection_t *conn = pn_session_connection(ssn); | |
| //Sweep thru every pn_link in this connection and a matching session and zero out the | |
| // qd_link->pn_link reference. We do this in order to not miss any pn_links | |
| pn_link = pn_link_head(conn, 0); | |
| while (pn_link) { | |
| if (pn_link_session(pn_link) == ssn) { | |
| qd_link_t *qd_link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if ((pn_link_state(pn_link) == (PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE))) { | |
| if (qd_link && qd_link->node) { | |
| if (qd_conn->policy_settings) { | |
| if (qd_link->direction == QD_OUTGOING) { | |
| qd_conn->n_receivers--; | |
| assert(qd_conn->n_receivers >= 0); | |
| } else { | |
| qd_conn->n_senders--; | |
| assert(qd_conn->n_senders >= 0); | |
| } | |
| } | |
| qd_log(container->log_source, QD_LOG_DEBUG, | |
| "Aborting link '%s' due to parent session end", | |
| pn_link_name(pn_link)); | |
| qd_link->node->ntype->link_detach_handler(qd_link->node->context, | |
| qd_link, QD_LOST); | |
| } | |
| } | |
| if (qd_link) | |
| qd_link->pn_link = 0; | |
| } | |
| pn_link = pn_link_next(pn_link, 0); | |
| } | |
| if (qd_conn->policy_settings) { | |
| qd_conn->n_sessions--; | |
| } | |
| pn_session_close(ssn); | |
| } | |
| else if (pn_session_state(ssn) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(ssn); | |
| qd_session_free(qd_ssn); | |
| add_session_to_free_list(&qd_conn->free_link_session_list, ssn); | |
| } | |
| } | |
| break; | |
| case PN_LINK_REMOTE_OPEN : | |
| if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) { | |
| pn_link = pn_event_link(event); | |
| if (pn_link_state(pn_link) & PN_LOCAL_UNINIT) { | |
| if (pn_link_is_sender(pn_link)) { | |
| if (qd_conn->policy_settings) { | |
| if (!qd_policy_approve_amqp_receiver_link(pn_link, qd_conn)) { | |
| break; | |
| } | |
| qd_conn->n_receivers++; | |
| } | |
| setup_outgoing_link(container, pn_link); | |
| } else { | |
| if (qd_conn->policy_settings) { | |
| if (!qd_policy_approve_amqp_sender_link(pn_link, qd_conn)) { | |
| break; | |
| } | |
| qd_conn->n_senders++; | |
| } | |
| setup_incoming_link(container, pn_link, qd_connection_max_message_size(qd_conn)); | |
| } | |
| } else if (pn_link_state(pn_link) & PN_LOCAL_ACTIVE) | |
| handle_link_open(container, pn_link); | |
| } | |
| break; | |
| case PN_LINK_REMOTE_DETACH : | |
| case PN_LINK_REMOTE_CLOSE : | |
| if (!(pn_connection_state(conn) & PN_LOCAL_CLOSED)) { | |
| pn_link = pn_event_link(event); | |
| qd_link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (qd_link) { | |
| qd_node_t *node = qd_link->node; | |
| qd_detach_type_t dt = pn_event_type(event) == PN_LINK_REMOTE_CLOSE ? QD_CLOSED : QD_DETACHED; | |
| if (!node && qd_link->pn_link == pn_link) { | |
| pn_link_close(pn_link); | |
| } | |
| if (qd_conn->policy_counted && qd_conn->policy_settings) { | |
| if (pn_link_is_sender(pn_link)) { | |
| qd_conn->n_receivers--; | |
| qd_log(container->log_source, QD_LOG_TRACE, | |
| "Closed receiver link %s. n_receivers: %d", | |
| pn_link_name(pn_link), qd_conn->n_receivers); | |
| assert (qd_conn->n_receivers >= 0); | |
| } else { | |
| qd_conn->n_senders--; | |
| qd_log(container->log_source, QD_LOG_TRACE, | |
| "Closed sender link %s. n_senders: %d", | |
| pn_link_name(pn_link), qd_conn->n_senders); | |
| assert (qd_conn->n_senders >= 0); | |
| } | |
| } | |
| if (pn_link_state(pn_link) & PN_LOCAL_CLOSED) { | |
| add_link_to_free_list(&qd_conn->free_link_session_list, pn_link); | |
| } | |
| if (node) { | |
| node->ntype->link_detach_handler(node->context, qd_link, dt); | |
| } | |
| } else { | |
| add_link_to_free_list(&qd_conn->free_link_session_list, pn_link); | |
| } | |
| } | |
| break; | |
| case PN_LINK_LOCAL_DETACH: | |
| case PN_LINK_LOCAL_CLOSE: | |
| pn_link = pn_event_link(event); | |
| if (pn_link_state(pn_link) == (PN_LOCAL_CLOSED | PN_REMOTE_CLOSED)) { | |
| add_link_to_free_list(&qd_conn->free_link_session_list, pn_link); | |
| } | |
| break; | |
| case PN_LINK_FLOW : | |
| pn_link = pn_event_link(event); | |
| qd_link = (qd_link_t*) pn_link_get_context(pn_link); | |
| if (qd_link && qd_link->node && qd_link->node->ntype->link_flow_handler) | |
| qd_link->node->ntype->link_flow_handler(qd_link->node->context, qd_link); | |
| break; | |
| case PN_DELIVERY : | |
| delivery = pn_event_delivery(event); | |
| pn_link = pn_event_link(event); | |
| if (pn_delivery_readable(delivery)) | |
| do_receive(pn_link, delivery); | |
| if (pn_delivery_updated(delivery) || pn_delivery_settled(delivery)) { | |
| do_updated(delivery); | |
| pn_delivery_clear(delivery); | |
| } | |
| break; | |
| case PN_CONNECTION_WAKE: | |
| if (!qd_conn->closed) | |
| writable_handler(container, conn, qd_conn); | |
| break; | |
| case PN_TRANSPORT_CLOSED: | |
| close_handler(container, conn, qd_conn); | |
| break; | |
| default: | |
| break; | |
| } | |
| } | |
| qd_container_t *qd_container(qd_dispatch_t *qd) | |
| { | |
| qd_container_t *container = NEW(qd_container_t); | |
| ZERO(container); | |
| 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->nodes); | |
| DEQ_INIT(container->node_type_list); | |
| qd_server_set_container(qd, container); | |
| qd_log(container->log_source, QD_LOG_TRACE, "Container Initialized"); | |
| return container; | |
| } | |
| void qd_container_free(qd_container_t *container) | |
| { | |
| if (!container) return; | |
| if (container->default_node) | |
| qd_container_destroy_node(container->default_node); | |
| qd_link_t *link = DEQ_HEAD(container->links); | |
| while (link) { | |
| DEQ_REMOVE_HEAD(container->links); | |
| cleanup_link(link); | |
| free_qd_link_t(link); | |
| link = DEQ_HEAD(container->links); | |
| } | |
| qd_node_t *node = DEQ_HEAD(container->nodes); | |
| while (node) { | |
| qd_container_destroy_node(node); | |
| node = DEQ_HEAD(container->nodes); | |
| } | |
| qdc_node_type_t *nt = DEQ_HEAD(container->node_type_list); | |
| while (nt) { | |
| DEQ_REMOVE_HEAD(container->node_type_list); | |
| free(nt); | |
| nt = DEQ_HEAD(container->node_type_list); | |
| } | |
| qd_hash_free(container->node_map); | |
| qd_hash_free(container->node_type_map); | |
| 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_iterator_t *iter = qd_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_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; | |
| DEQ_ITEM_INIT(node); | |
| 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_iterator_t *iter = qd_iterator_string(name, ITER_VIEW_ALL); | |
| sys_mutex_lock(container->lock); | |
| result = qd_hash_insert(container->node_map, iter, node, 0); | |
| if (result >= 0) | |
| DEQ_INSERT_HEAD(container->nodes, node); | |
| sys_mutex_unlock(container->lock); | |
| qd_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_iterator_t *iter = qd_iterator_string(node->name, ITER_VIEW_ALL); | |
| sys_mutex_lock(container->lock); | |
| qd_hash_remove(container->node_map, iter); | |
| DEQ_REMOVE(container->nodes, node); | |
| sys_mutex_unlock(container->lock); | |
| qd_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, qd_session_class_t ssn_class) | |
| { | |
| const qd_server_config_t * cf = qd_connection_config(conn); | |
| pn_session_t *pn_ssn = conn->pn_sessions[ssn_class]; | |
| if (!pn_ssn) { | |
| pn_ssn = pn_session(qd_connection_pn(conn)); | |
| if (!pn_ssn) { | |
| return NULL; | |
| } | |
| qd_session_t *qd_ssn = qd_session(pn_ssn); | |
| if (!qd_ssn) { | |
| pn_session_free(pn_ssn); | |
| return NULL; // PN_LOCAL_CLOSE event will free pn_ssn | |
| } | |
| conn->pn_sessions[ssn_class] = pn_ssn; | |
| pn_session_set_incoming_capacity(pn_ssn, cf->incoming_capacity); | |
| pn_session_open(pn_ssn); | |
| } | |
| qd_link_t *link = new_qd_link_t(); | |
| if (!link) { | |
| return NULL; // ok to keep qd_session around - will reuse & free on conn close | |
| } | |
| ZERO(link); | |
| sys_mutex_lock(node->container->lock); | |
| DEQ_INSERT_TAIL(node->container->links, link); | |
| sys_mutex_unlock(node->container->lock); | |
| link->pn_sess = pn_ssn; | |
| if (dir == QD_OUTGOING) | |
| link->pn_link = pn_sender(link->pn_sess, name); | |
| else | |
| link->pn_link = pn_receiver(link->pn_sess, name); | |
| link->direction = dir; | |
| link->context = node->context; | |
| link->node = node; | |
| link->remote_snd_settle_mode = pn_link_remote_snd_settle_mode(link->pn_link); | |
| pn_link_set_context(link->pn_link, link); | |
| return link; | |
| } | |
| void qd_link_free(qd_link_t *link) | |
| { | |
| if (!link) return; | |
| qd_container_t *container = link->node->container; | |
| sys_mutex_lock(container->lock); | |
| DEQ_REMOVE(container->links, link); | |
| sys_mutex_unlock(container->lock); | |
| qd_node_t *node = link->node; | |
| node->ntype->link_abandoned_deliveries_handler(node->context, link); | |
| cleanup_link(link); | |
| free_qd_link_t(link); | |
| } | |
| qd_link_ref_list_t *qd_link_get_ref_list(qd_link_t *link) | |
| { | |
| return &link->ref_list; | |
| } | |
| 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; | |
| } | |
| pn_link_t *qd_link_pn(qd_link_t *link) | |
| { | |
| return link->pn_link; | |
| } | |
| pn_session_t *qd_link_pn_session(qd_link_t *link) | |
| { | |
| return link->pn_sess; | |
| } | |
| bool qd_link_is_q2_limit_unbounded(qd_link_t *link) | |
| { | |
| return link->q2_limit_unbounded; | |
| } | |
| void qd_link_set_q2_limit_unbounded(qd_link_t *link, bool q2_limit_unbounded) | |
| { | |
| link->q2_limit_unbounded = q2_limit_unbounded; | |
| } | |
| qd_direction_t qd_link_direction(const qd_link_t *link) | |
| { | |
| return link->direction; | |
| } | |
| pn_snd_settle_mode_t qd_link_remote_snd_settle_mode(const qd_link_t *link) | |
| { | |
| return link->remote_snd_settle_mode; | |
| } | |
| 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 || !ctx->opened) | |
| 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_close(qd_link_t *link) | |
| { | |
| if (link->pn_link) | |
| pn_link_close(link->pn_link); | |
| } | |
| void qd_link_detach(qd_link_t *link) | |
| { | |
| if (link->pn_link) { | |
| pn_link_detach(link->pn_link); | |
| pn_link_close(link->pn_link); | |
| } | |
| } | |
| void *qd_link_get_node_context(const qd_link_t *link) | |
| { | |
| return (link && link->node) ? link->node->context : 0; | |
| } | |
| /** sending link has entered Q3 flow control */ | |
| void qd_link_q3_block(qd_link_t *link) | |
| { | |
| assert(link); | |
| if (!link->q3_blocked && link->pn_sess) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(link->pn_sess); | |
| assert(qd_ssn); | |
| link->q3_blocked = true; | |
| DEQ_INSERT_TAIL_N(Q3, qd_ssn->q3_blocked_links, link); | |
| } | |
| } | |
| void qd_link_q3_unblock(qd_link_t *link) | |
| { | |
| assert(link); | |
| if (link->q3_blocked) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(link->pn_sess); | |
| assert(qd_ssn); | |
| DEQ_REMOVE_N(Q3, qd_ssn->q3_blocked_links, link); | |
| link->q3_blocked = false; | |
| } | |
| } | |
| uint64_t qd_link_link_id(const qd_link_t *link) | |
| { | |
| return link->link_id; | |
| } | |
| void qd_link_set_link_id(qd_link_t *link, uint64_t link_id) | |
| { | |
| link->link_id = link_id; | |
| } | |
| qd_session_t *qd_session(pn_session_t *pn_ssn) | |
| { | |
| assert(pn_ssn); | |
| qd_session_t *qd_ssn = qd_session_from_pn(pn_ssn); | |
| if (!qd_ssn) { | |
| qd_ssn = new_qd_session_t(); | |
| if (qd_ssn) { | |
| ZERO(qd_ssn); | |
| qd_ssn->pn_sess = pn_ssn; | |
| DEQ_INIT(qd_ssn->q3_blocked_links); | |
| qd_session_link_pn(qd_ssn, pn_ssn); | |
| } | |
| } | |
| return qd_ssn; | |
| } | |
| void qd_session_free(qd_session_t *qd_ssn) | |
| { | |
| if (qd_ssn) { | |
| qd_link_t *link = DEQ_HEAD(qd_ssn->q3_blocked_links); | |
| while (link) { | |
| qd_link_q3_unblock(link); // removes link from list | |
| link = DEQ_HEAD(qd_ssn->q3_blocked_links); | |
| } | |
| if (qd_ssn->pn_sess) { | |
| qd_session_unlink_pn(qd_ssn, qd_ssn->pn_sess); | |
| } | |
| free_qd_session_t(qd_ssn); | |
| } | |
| } | |
| qd_link_list_t *qd_session_q3_blocked_links(qd_session_t *qd_ssn) | |
| { | |
| return qd_ssn ? &qd_ssn->q3_blocked_links : 0; | |
| } | |
| bool qd_session_is_q3_blocked(const qd_session_t *qd_ssn) | |
| { | |
| return qd_ssn && !DEQ_IS_EMPTY(qd_ssn->q3_blocked_links); | |
| } | |
| /** release all qd_session_t instances for the connection | |
| * called prior to releasing the qd_connection_t | |
| */ | |
| void qd_session_cleanup(qd_connection_t *qd_conn) | |
| { | |
| pn_connection_t *pn_conn = (qd_conn) ? qd_conn->pn_conn : 0; | |
| if (!pn_conn) | |
| return; | |
| pn_session_t *pn_ssn = pn_session_head(pn_conn, 0); | |
| while (pn_ssn) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(pn_ssn); | |
| qd_session_free(qd_ssn); | |
| pn_ssn = pn_session_next(pn_ssn, 0); | |
| } | |
| } | |
| void qd_link_set_incoming_msg(qd_link_t *link, qd_message_t *msg) | |
| { | |
| if (msg) { | |
| qd_alloc_set_safe_ptr(&link->incoming_msg, msg); | |
| } else { | |
| qd_nullify_safe_ptr(&link->incoming_msg); | |
| } | |
| } |