| /* |
| * 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); |
| } |
| } |
