| /* | |
| * 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 "delivery.h" | |
| #include "dispatch_private.h" | |
| #include "policy.h" | |
| #include "router_private.h" | |
| #include "qpid/dispatch.h" | |
| #include "qpid/dispatch/protocol_adaptor.h" | |
| #include "qpid/dispatch/proton_utils.h" | |
| #include <proton/sasl.h> | |
| #include <inttypes.h> | |
| #include <stdbool.h> | |
| #include <stdio.h> | |
| #include <stdlib.h> | |
| #include <string.h> | |
| const char *QD_ROUTER_NODE_TYPE = "router.node"; | |
| const char *QD_ROUTER_ADDRESS_TYPE = "router.address"; | |
| const char *QD_ROUTER_LINK_TYPE = "router.link"; | |
| static qdr_protocol_adaptor_t *amqp_direct_adaptor = 0; | |
| static char *router_role = "inter-router"; | |
| static char *container_role = "route-container"; | |
| static char *edge_role = "edge"; | |
| static char *direct_prefix; | |
| static char *node_id; | |
| static uint8_t *encoded_node_id; | |
| static size_t encoded_node_id_len; | |
| static void deferred_AMQP_rx_handler(void *context, bool discard); | |
| static bool parse_failover_property_list(qd_router_t *router, qd_connection_t *conn, pn_data_t *props); | |
| const char *QD_AMQP_COND_OVERSIZE_DESCRIPTION = "Message size exceeded"; | |
| //============================================================================== | |
| // Functions to handle the linkage between proton deliveries and qdr deliveries | |
| //============================================================================== | |
| // | |
| // qd_link.list_of_references(pn_delivery_t) | |
| // pn_delivery.context => reference-entry | |
| // qdr_delivery.context => pn_delivery | |
| // | |
| // reject a delivery, setting the apropriate condition fields so the sender can | |
| // determine the reason the message was rejected | |
| // | |
| static inline void _reject_delivery(pn_delivery_t *pnd, const char *error_name, const char *description) | |
| { | |
| assert(error_name && description); | |
| pn_condition_t *lcond = pn_disposition_condition(pn_delivery_local(pnd)); | |
| (void) pn_condition_set_name(lcond, error_name); | |
| (void) pn_condition_set_description(lcond, description); | |
| pn_delivery_update(pnd, PN_REJECTED); | |
| } | |
| static inline const char *_get_tenant_space(qd_connection_t *conn, int *length) | |
| { | |
| qdr_connection_t *qdr_conn = (qdr_connection_t*) qd_connection_get_context(conn); | |
| return qdr_connection_get_tenant_space(qdr_conn, length); | |
| } | |
| static void qdr_node_connect_deliveries(qd_link_t *link, qdr_delivery_t *qdlv, pn_delivery_t *pdlv) | |
| { | |
| qd_link_ref_t *ref = new_qd_link_ref_t(); | |
| qd_link_ref_list_t *list = qd_link_get_ref_list(link); | |
| ZERO(ref); | |
| ref->ref = qdlv; | |
| DEQ_INSERT_TAIL(*list, ref); | |
| pn_delivery_set_context(pdlv, ref); | |
| qdr_delivery_set_context(qdlv, pdlv); | |
| qdr_delivery_incref(qdlv, "referenced by a pn_delivery"); | |
| } | |
| static void qdr_node_disconnect_deliveries(qdr_core_t *core, qd_link_t *link, qdr_delivery_t *qdlv, pn_delivery_t *pdlv) | |
| { | |
| if (!link) | |
| return; | |
| qd_link_ref_t *ref = (qd_link_ref_t*) pn_delivery_get_context(pdlv); | |
| qd_link_ref_list_t *list = qd_link_get_ref_list(link); | |
| if (ref) { | |
| DEQ_REMOVE(*list, ref); | |
| free_qd_link_ref_t(ref); | |
| pn_delivery_set_context(pdlv, 0); | |
| qdr_delivery_set_context(qdlv, 0); | |
| qdr_delivery_decref(core, qdlv, "removed reference from pn_delivery"); | |
| } | |
| } | |
| static pn_delivery_t *qdr_node_delivery_pn_from_qdr(qdr_delivery_t *dlv) | |
| { | |
| return dlv ? (pn_delivery_t*) qdr_delivery_get_context(dlv) : 0; | |
| } | |
| static qdr_delivery_t *qdr_node_delivery_qdr_from_pn(pn_delivery_t *dlv) | |
| { | |
| qd_link_ref_t *ref = (qd_link_ref_t*) pn_delivery_get_context(dlv); | |
| return ref ? (qdr_delivery_t*) ref->ref : 0; | |
| } | |
| void qd_link_abandoned_deliveries_handler(void *context, qd_link_t *link) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qd_link_ref_list_t *list = qd_link_get_ref_list(link); | |
| qd_link_ref_t *ref = DEQ_HEAD(*list); | |
| while (ref) { | |
| DEQ_REMOVE_HEAD(*list); | |
| qdr_delivery_t *dlv = (qdr_delivery_t*) ref->ref; | |
| ref->ref = 0; | |
| qdr_delivery_set_context(dlv, 0); | |
| qdr_delivery_decref(router->router_core, dlv, "qd_link_abandoned_deliveries_handler"); | |
| free_qd_link_ref_t(ref); | |
| ref = DEQ_HEAD(*list); | |
| } | |
| } | |
| // read the delivery-state set by the remote endpoint | |
| // | |
| static qd_delivery_state_t *qd_delivery_read_remote_state(pn_delivery_t *pnd) | |
| { | |
| qd_delivery_state_t *dstate = 0; | |
| uint64_t outcome = pn_delivery_remote_state(pnd); | |
| switch (outcome) { | |
| case 0: | |
| // not set - no delivery-state | |
| break; | |
| case PN_RECEIVED: { | |
| pn_disposition_t *disp = pn_delivery_remote(pnd); | |
| dstate = qd_delivery_state(); | |
| dstate->section_number = pn_disposition_get_section_number(disp); | |
| dstate->section_offset = pn_disposition_get_section_offset(disp); | |
| break; | |
| } | |
| case PN_ACCEPTED: | |
| case PN_RELEASED: | |
| // no associated state (that we care about) | |
| break; | |
| case PN_REJECTED: { | |
| // See AMQP 1.0 section 3.4.4 Rejected | |
| pn_condition_t *cond = pn_disposition_condition(pn_delivery_remote(pnd)); | |
| dstate = qd_delivery_state(); | |
| dstate->error = qdr_error_from_pn(cond); | |
| break; | |
| } | |
| case PN_MODIFIED: { | |
| // See AMQP 1.0 section 3.4.5 Modified | |
| pn_disposition_t *disp = pn_delivery_remote(pnd); | |
| bool failed = pn_disposition_is_failed(disp); | |
| bool undeliverable = pn_disposition_is_undeliverable(disp); | |
| pn_data_t *anno = pn_disposition_annotations(disp); | |
| // avoid expensive alloc if only default values found | |
| const bool need_anno = (anno && pn_data_size(anno) > 0); | |
| if (failed || undeliverable || need_anno) { | |
| dstate = qd_delivery_state(); | |
| dstate->delivery_failed = failed; | |
| dstate->undeliverable_here = undeliverable; | |
| if (need_anno) { | |
| dstate->annotations = pn_data(0); | |
| pn_data_copy(dstate->annotations, anno); | |
| } | |
| } | |
| break; | |
| } | |
| default: { | |
| // Check for custom state data. Custom outcomes and AMQP 1.0 | |
| // Transaction defined outcomes will all be numerically > | |
| // PN_MODIFIED. See Part 4: Transactions and section 1.5 Descriptor | |
| // Values in the AMQP 1.0 spec. | |
| if (outcome > PN_MODIFIED) { | |
| pn_data_t *data = pn_disposition_data(pn_delivery_remote(pnd)); | |
| if (data && pn_data_size(data) > 0) { | |
| dstate = qd_delivery_state(); | |
| dstate->extension = pn_data(0); | |
| pn_data_copy(dstate->extension, data); | |
| } | |
| } | |
| break; | |
| } | |
| } // end switch | |
| return dstate; | |
| } | |
| // Set the delivery-state to be sent to the remote endpoint. | |
| // | |
| static void qd_delivery_write_local_state(pn_delivery_t *pnd, uint64_t outcome, const qd_delivery_state_t *dstate) | |
| { | |
| if (pnd && dstate) { | |
| switch (outcome) { | |
| case PN_RECEIVED: { | |
| pn_disposition_t *ldisp = pn_delivery_local(pnd); | |
| pn_disposition_set_section_number(ldisp, dstate->section_number); | |
| pn_disposition_set_section_offset(ldisp, dstate->section_offset); | |
| break; | |
| } | |
| case PN_ACCEPTED: | |
| case PN_RELEASED: | |
| // no associated state (that we care about) | |
| break; | |
| case PN_REJECTED: | |
| if (dstate->error) { | |
| pn_condition_t *condition = pn_disposition_condition(pn_delivery_local(pnd)); | |
| char *name = qdr_error_name(dstate->error); | |
| char *description = qdr_error_description(dstate->error); | |
| pn_condition_set_name(condition, (const char*) name); | |
| pn_condition_set_description(condition, (const char*) description); | |
| if (qdr_error_info(dstate->error)) { | |
| pn_data_copy(pn_condition_info(condition), qdr_error_info(dstate->error)); | |
| } | |
| //proton makes copies of name and description, so it is ok to free them here. | |
| free(name); | |
| free(description); | |
| } | |
| break; | |
| case PN_MODIFIED: { | |
| pn_disposition_t *ldisp = pn_delivery_local(pnd); | |
| pn_disposition_set_failed(ldisp, dstate->delivery_failed); | |
| pn_disposition_set_undeliverable(ldisp, dstate->undeliverable_here); | |
| if (dstate->annotations) | |
| pn_data_copy(pn_disposition_annotations(ldisp), dstate->annotations); | |
| break; | |
| } | |
| default: | |
| if (dstate->extension) | |
| pn_data_copy(pn_disposition_data(pn_delivery_local(pnd)), dstate->extension); | |
| break; | |
| } | |
| } | |
| } | |
| /** | |
| * Determine the role of a connection | |
| */ | |
| static void qd_router_connection_get_config(const qd_connection_t *conn, | |
| qdr_connection_role_t *role, | |
| int *cost, | |
| const char **name, | |
| bool *multi_tenant, | |
| bool *strip_annotations_in, | |
| bool *strip_annotations_out, | |
| int *link_capacity) | |
| { | |
| if (conn) { | |
| const qd_server_config_t *cf = qd_connection_config(conn); | |
| *strip_annotations_in = cf ? cf->strip_inbound_annotations : true; | |
| *strip_annotations_out = cf ? cf->strip_outbound_annotations : true; | |
| *link_capacity = cf ? cf->link_capacity : 1; | |
| if (cf && (strcmp(cf->role, router_role) == 0)) { | |
| *strip_annotations_in = false; | |
| *strip_annotations_out = false; | |
| *role = QDR_ROLE_INTER_ROUTER; | |
| *cost = cf->inter_router_cost; | |
| } else if (cf && (strcmp(cf->role, edge_role) == 0)) { | |
| *strip_annotations_in = false; | |
| *strip_annotations_out = false; | |
| *role = QDR_ROLE_EDGE_CONNECTION; | |
| *cost = cf->inter_router_cost; | |
| } else if (cf && (strcmp(cf->role, container_role) == 0)) // backward compat | |
| *role = QDR_ROLE_ROUTE_CONTAINER; | |
| else | |
| *role = QDR_ROLE_NORMAL; | |
| *name = cf ? cf->name : 0; | |
| if (*name) { | |
| if (strncmp("listener/", *name, 9) == 0 || | |
| strncmp("connector/", *name, 10) == 0) | |
| *name = 0; | |
| } | |
| *multi_tenant = cf ? cf->multi_tenant : false; | |
| } | |
| } | |
| static int AMQP_writable_conn_handler(void *type_context, qd_connection_t *conn, void *context) | |
| { | |
| qdr_connection_t *qconn = (qdr_connection_t*) qd_connection_get_context(conn); | |
| if (qconn) | |
| return qdr_connection_process(qconn); | |
| return 0; | |
| } | |
| static qd_iterator_t *process_router_annotations(qd_router_t *router, | |
| qd_message_t *msg, | |
| qd_bitmask_t **link_exclusions, | |
| uint32_t *distance, | |
| int *ingress_index) | |
| { | |
| qd_iterator_t *ingress_iter = 0; | |
| bool edge_mode = router->router_mode == QD_ROUTER_MODE_EDGE; | |
| *link_exclusions = 0; | |
| *distance = 0; | |
| *ingress_index = 0; | |
| if (!edge_mode) { // Edge routers do not use trace or ingress meta-data | |
| qd_parsed_field_t *trace = qd_message_get_trace(msg); | |
| if (trace && qd_parse_is_list(trace)) { | |
| // | |
| // Return the distance in hops that this delivery has traveled. | |
| // | |
| *distance = qd_parse_sub_count(trace); | |
| // | |
| // Create a link-exclusion map for the items in the trace. This map will | |
| // contain a one-bit for each link that leads to a neighbor router that | |
| // the message has already passed through. | |
| // | |
| *link_exclusions = qd_tracemask_create(router->tracemask, trace, ingress_index); | |
| } | |
| qd_parsed_field_t *ingress = qd_message_get_ingress_router(msg); | |
| if (ingress && qd_parse_is_scalar(ingress)) { | |
| ingress_iter = qd_parse_raw(ingress); | |
| } | |
| } else { | |
| // Edge routers do not propagate trace or ingress | |
| qd_message_disable_trace_annotation(msg); | |
| qd_message_disable_ingress_router_annotation(msg); | |
| } | |
| // | |
| // Return the iterator to the ingress field _if_ it was present. | |
| // Otherwise this router is the ingress - return NULL. | |
| // | |
| return ingress_iter; | |
| } | |
| static void log_link_message(qd_connection_t *conn, pn_link_t *pn_link, qd_message_t *msg) | |
| { | |
| assert(conn && pn_link && msg); | |
| qd_log_source_t *logger = qd_message_log_source(); | |
| // the message processing is expensive as this is done for every message received. | |
| // Do not bother if not tracing. | |
| if (qd_log_enabled(logger, QD_LOG_TRACE)) { | |
| const qd_server_config_t *cf = qd_connection_config(conn); | |
| if (!cf) return; | |
| size_t repr_len = qd_message_repr_len(); | |
| char *buf = qd_malloc(repr_len); | |
| const char *msg_str = qd_message_oversize(msg) ? "oversize message" : | |
| qd_message_aborted(msg) ? "aborted message" : | |
| qd_message_repr(msg, buf, repr_len, cf->log_bits); | |
| if (msg_str) { | |
| const char *src = pn_terminus_get_address(pn_link_source(pn_link)); | |
| const char *tgt = pn_terminus_get_address(pn_link_target(pn_link)); | |
| qd_log(logger, QD_LOG_TRACE, | |
| "[C%"PRIu64"]: %s %s on link '%s' (%s -> %s)", | |
| qd_connection_connection_id(conn), | |
| pn_link_is_sender(pn_link) ? "Sent" : "Received", | |
| msg_str, | |
| pn_link_name(pn_link), | |
| src ? src : "", | |
| tgt ? tgt : ""); | |
| } | |
| free(buf); | |
| } | |
| } | |
| /** | |
| * Inbound Delivery Handler | |
| * | |
| * @return true if we've advanced to the next delivery on this link and it is | |
| * ready for rx processing. This will cause the container to immediately | |
| * re-call this function with the next delivery. | |
| */ | |
| static bool AMQP_rx_handler(void* context, qd_link_t *link) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| pn_link_t *pn_link = qd_link_pn(link); | |
| assert(pn_link); | |
| if (!pn_link) | |
| return false; | |
| // ensure the current delivery is readable | |
| pn_delivery_t *pnd = pn_link_current(pn_link); | |
| if (!pnd) | |
| return false; | |
| qd_connection_t *conn = qd_link_connection(link); | |
| // DISPATCH-1628 DISPATCH-975 exit if router already closed this connection | |
| if (conn->closed_locally) { | |
| return false; | |
| } | |
| qdr_delivery_t *delivery = qdr_node_delivery_qdr_from_pn(pnd); | |
| bool next_delivery = false; | |
| // | |
| // Receive the message into a local representation. | |
| // | |
| qd_message_t *msg = qd_message_receive(pnd); | |
| bool receive_complete = qd_message_receive_complete(msg); | |
| if (!qd_message_oversize(msg)) { | |
| // message not rejected as oversize | |
| if (receive_complete) { | |
| // | |
| // The entire message has been received and we are ready to consume the delivery by calling pn_link_advance(). | |
| // | |
| pn_link_advance(pn_link); | |
| next_delivery = pn_link_current(pn_link) != 0; | |
| } | |
| if (qd_message_is_discard(msg)) { | |
| // | |
| // Message has been marked for discard, no further processing necessary | |
| // | |
| if (receive_complete) { | |
| // If this discarded delivery has already been settled by proton, | |
| // set the presettled flag on the delivery to true if it is not already true. | |
| // Since the entire message has already been received, we directly call the | |
| // function to set the pre-settled flag since we cannot go thru the core-thread | |
| // to do this since the delivery has been discarded. | |
| // Discarded streaming deliveries are not put thru the core thread via the continue action. | |
| if (pn_delivery_settled(pnd)) | |
| qdr_delivery_set_presettled(delivery); | |
| uint64_t local_disp = qdr_delivery_disposition(delivery); | |
| // | |
| // Call pn_delivery_update only if the local disposition is different than the pn_delivery's local disposition. | |
| // This will make sure we call pn_delivery_update only when necessary. | |
| // | |
| if (local_disp != 0 && local_disp != pn_delivery_local_state(pnd)) { | |
| // | |
| // DISPATCH-1626 - This enables pn_delivery_update() and pn_delivery_settle() to be called back to back in the same function call. | |
| // CORE_delivery_update() will handle most of the other cases where we need to call pn_delivery_update() followed by pn_delivery_settle(). | |
| // | |
| pn_delivery_update(pnd, local_disp); | |
| } | |
| // note: expected that the code that set discard has handled | |
| // setting disposition and updating flow! | |
| pn_delivery_settle(pnd); | |
| if (delivery) { | |
| // if delivery already exists then the core thread discarded this | |
| // delivery, it will eventually free the qdr_delivery_t and its | |
| // associated message - do not free it here. | |
| qdr_node_disconnect_deliveries(router->router_core, link, delivery, pnd); | |
| } else { | |
| qd_message_free(msg); | |
| } | |
| } | |
| return next_delivery; | |
| } | |
| } else { | |
| // message is oversize | |
| if (receive_complete) { | |
| // set condition, reject, and settle the incoming delivery | |
| _reject_delivery(pnd, QD_AMQP_COND_MESSAGE_SIZE_EXCEEDED, QD_AMQP_COND_OVERSIZE_DESCRIPTION); | |
| pn_delivery_settle(pnd); | |
| // close the link | |
| pn_link_close(pn_link); | |
| // set condition and close the connection | |
| pn_connection_t * pn_conn = qd_connection_pn(conn); | |
| pn_condition_t * cond = pn_connection_condition(pn_conn); | |
| (void) pn_condition_set_name( cond, QD_AMQP_COND_CONNECTION_FORCED); | |
| (void) pn_condition_set_description(cond, QD_AMQP_COND_OVERSIZE_DESCRIPTION); | |
| pn_connection_close(pn_conn); | |
| if (!delivery) { | |
| // this message has not been forwarded yet, so it will not be | |
| // cleaned up when the link is freed. | |
| qd_message_free(msg); | |
| } | |
| // stop all message reception on this connection | |
| conn->closed_locally = true; | |
| } | |
| return false; | |
| // oversize messages are not processed any further | |
| } | |
| // | |
| // If the delivery already exists we've already passed it to the core (2nd | |
| // frame for a multi-frame transfer). Simply continue. | |
| // | |
| if (delivery) { | |
| qdr_delivery_continue(router->router_core, delivery, pn_delivery_settled(pnd)); | |
| return next_delivery; | |
| } | |
| // | |
| // No pre-existing delivery means we're starting a new delivery or | |
| // continuing a delivery that has not accumulated enough of the message | |
| // for forwarding. | |
| // | |
| qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link); | |
| if (!rlink) { | |
| // receive link was closed or deleted - can't be forwarded | |
| // so no use setting disposition or adding flow | |
| qd_message_set_discard(msg, true); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| return next_delivery; | |
| } | |
| // | |
| // Validate the content of the delivery as an AMQP message. This is done | |
| // partially, only to validate that we can find the fields we need to route | |
| // the message. | |
| // | |
| // If per-message tracing is configured then validate the sections | |
| // necessary for logging. | |
| // | |
| // link-routing: it is not necessary to validate any sections, but doing so | |
| // will force a message validity check and ensure the message is not null. | |
| // | |
| // If the link is anonymous, we must validate through the message | |
| // properties to find the 'to' field. If the link is not anonymous, we | |
| // don't need the 'to' field as we will be using the address from the link | |
| // target. | |
| // | |
| // Check if the user id needs to be validated (see below). If it does we | |
| // need to validate the message properties section. | |
| // | |
| // Otherwise check the message annotations for router annotations necessary | |
| // for forwarding. | |
| // | |
| const bool link_routed = qdr_link_is_routed(rlink); | |
| const bool anonymous_link = qdr_link_is_anonymous(rlink); | |
| const bool check_user = (conn->policy_settings && !conn->policy_settings->spec.allowUserIdProxy); | |
| const qd_server_config_t *cf = qd_connection_config(conn); | |
| const qd_message_depth_t depth = (cf && cf->log_bits != 0) ? QD_DEPTH_APPLICATION_PROPERTIES | |
| : (link_routed) ? QD_DEPTH_HEADER | |
| : (anonymous_link || check_user) ? QD_DEPTH_PROPERTIES | |
| : QD_DEPTH_MESSAGE_ANNOTATIONS; | |
| const qd_message_depth_status_t depth_valid = qd_message_check_depth(msg, depth); | |
| switch (depth_valid) { | |
| case QD_MESSAGE_DEPTH_INVALID: | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "[C%"PRIu64"][L%"PRIu64"] Incoming message validation failed - rejected", | |
| conn->connection_id, | |
| qd_link_link_id(link)); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| _reject_delivery(pnd, QD_AMQP_COND_DECODE_ERROR, "invalid message format"); | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| return next_delivery; | |
| case QD_MESSAGE_DEPTH_INCOMPLETE: | |
| return false; // stop rx processing | |
| case QD_MESSAGE_DEPTH_OK: | |
| break; | |
| } | |
| // Handle the link-routed case | |
| // | |
| if (link_routed) { | |
| pn_delivery_tag_t dtag = pn_delivery_tag(pnd); | |
| if (dtag.size > QDR_DELIVERY_TAG_MAX) { | |
| qd_log(router->log_source, QD_LOG_DEBUG, "link route delivery failure: msg tag size exceeded %zd (max=%d)", | |
| dtag.size, QDR_DELIVERY_TAG_MAX); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| _reject_delivery(pnd, QD_AMQP_COND_INVALID_FIELD, "delivery tag length exceeded"); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| return next_delivery; | |
| } | |
| log_link_message(conn, pn_link, msg); | |
| delivery = qdr_link_deliver_to_routed_link(rlink, | |
| msg, | |
| pn_delivery_settled(pnd), | |
| (uint8_t*) dtag.start, | |
| dtag.size, | |
| pn_delivery_remote_state(pnd), | |
| qd_delivery_read_remote_state(pnd)); | |
| qd_link_set_incoming_msg(link, (qd_message_t*) 0); // msg no longer exclusive to qd_link | |
| qdr_node_connect_deliveries(link, delivery, pnd); | |
| qdr_delivery_decref(router->router_core, delivery, "release protection of return from deliver_to_routed_link"); | |
| return next_delivery; | |
| } | |
| // Determine if the user of this connection is allowed to proxy the user_id | |
| // of messages. A message user_id is proxied when the value in the message | |
| // properties section differs from the authenticated user name of the | |
| // connection. If the user is not allowed to proxy the user_id then the | |
| // message user_id must be blank or it must be equal to the connection user | |
| // name. | |
| // | |
| if (check_user) { | |
| // This connection must not allow proxied user_id | |
| qd_iterator_t *userid_iter = qd_message_field_iterator(msg, QD_FIELD_USER_ID); | |
| if (userid_iter) { | |
| // The user_id property has been specified | |
| if (qd_iterator_remaining(userid_iter) > 0) { | |
| // user_id property in message is not blank | |
| if (!qd_iterator_equal(userid_iter, (const unsigned char *)conn->user_id)) { | |
| // This message is rejected: attempted user proxy is disallowed | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "[C%"PRIu64"][L%"PRIu64"] Message rejected due to user_id proxy violation. User:%s", | |
| conn->connection_id, | |
| qd_link_link_id(link), | |
| conn->user_id); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| _reject_delivery(pnd, QD_AMQP_COND_UNAUTHORIZED_ACCESS, "user_id proxy violation"); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| qd_iterator_free(userid_iter); | |
| return next_delivery; | |
| } | |
| } | |
| qd_iterator_free(userid_iter); | |
| } | |
| } | |
| const char *ma_error = qd_message_parse_annotations(msg); | |
| if (ma_error) { | |
| qd_log(router->log_source, QD_LOG_WARNING, | |
| "[C%"PRIu64"][L%"PRIu64"] Message rejected - invalid MA section: %s", | |
| conn->connection_id, qd_link_link_id(link), ma_error); | |
| pn_condition_t *condition = pn_disposition_condition(pn_delivery_local(pnd)); | |
| pn_condition_set_name(condition, "amqp:invalid-field"); | |
| pn_condition_set_description(condition, ma_error); | |
| pn_delivery_update(pnd, PN_REJECTED); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| return next_delivery; | |
| } | |
| // | |
| // Head of line blocking avoidance (DISPATCH-1545) | |
| // | |
| // Before we can forward a message we need to determine whether or not this | |
| // message is "streaming" - a large message that has the potential to block | |
| // other messages sharing the trunk link. At this point we cannot for sure | |
| // know the actual length of the incoming message, so we employ the | |
| // following heuristic to determine if the message is "streaming": | |
| // | |
| // - If the message is receive-complete it is NOT a streaming message. | |
| // - If it is NOT receive-complete: | |
| // Continue buffering incoming data until: | |
| // - receive has completed => NOT a streaming message | |
| // - not rx-complete AND Q2 threshold hit => a streaming message | |
| // | |
| // Once Q2 is hit we MUST forward the message regardless of rx-complete | |
| // since Q2 will block forever unless the incoming data is drained via | |
| // forwarding. | |
| // | |
| if (!receive_complete) { | |
| if (qd_message_is_streaming(msg) || qd_message_is_Q2_blocked(msg)) { | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "[C%"PRIu64"][L%"PRIu64"] Incoming message classified as streaming. User:%s", | |
| conn->connection_id, | |
| qd_link_link_id(link), | |
| conn->user_id); | |
| } else { | |
| // Continue buffering this message | |
| return false; | |
| } | |
| } | |
| uint32_t distance = 0; | |
| int ingress_index = 0; // Default to _this_ router | |
| qd_bitmask_t *link_exclusions = 0; | |
| qd_iterator_t *ingress_iter = process_router_annotations(router, msg, &link_exclusions, &distance, &ingress_index); | |
| // | |
| // If this delivery has traveled further than the known radius of the network topology (plus 1), | |
| // release and settle the delivery. This can happen in the case of "flood" multicast where the | |
| // deliveries follow all available paths. This will only discard messages that will reach their | |
| // destinations via shorter paths. | |
| // | |
| if (distance > (router->topology_radius + 1)) { | |
| qd_bitmask_free(link_exclusions); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| pn_delivery_update(pnd, PN_RELEASED); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| return next_delivery; | |
| } | |
| if (anonymous_link) { | |
| qd_iterator_t *addr_iter = 0; | |
| int phase = 0; | |
| // | |
| // If the message has delivery annotations, get the to-override field from the annotations. | |
| // | |
| qd_parsed_field_t *ma_to = qd_message_get_to_override(msg); | |
| if (ma_to) { | |
| addr_iter = qd_iterator_dup(qd_parse_raw(ma_to)); | |
| phase = qd_message_get_phase_annotation(msg); | |
| } | |
| // | |
| // Still no destination address? Use the TO field from the message properties. | |
| // | |
| if (!addr_iter) { | |
| addr_iter = qd_message_field_iterator(msg, QD_FIELD_TO); | |
| // | |
| // If the address came from the TO field and we need to apply a tenant-space, | |
| // set the to-override with the annotated address. | |
| // | |
| if (addr_iter) { | |
| int tenant_space_length; | |
| const char *tenant_space = _get_tenant_space(conn, &tenant_space_length); | |
| if (tenant_space) { | |
| qd_iterator_reset_view(addr_iter, ITER_VIEW_ADDRESS_WITH_SPACE); | |
| qd_iterator_annotate_space(addr_iter, tenant_space, tenant_space_length); | |
| char *iter_str = (char *)qd_iterator_copy(addr_iter); | |
| qd_message_set_to_override_annotation(msg, iter_str); | |
| free(iter_str); | |
| } | |
| } | |
| } | |
| if (addr_iter) { | |
| if (!conn->policy_settings || qd_policy_approve_message_target(addr_iter, conn)) { | |
| qd_iterator_reset_view(addr_iter, ITER_VIEW_ADDRESS_HASH); | |
| if (phase > 0) | |
| qd_iterator_annotate_phase(addr_iter, '0' + (char) phase); | |
| log_link_message(conn, pn_link, msg); | |
| delivery = qdr_link_deliver_to(rlink, msg, ingress_iter, addr_iter, pn_delivery_settled(pnd), | |
| link_exclusions, ingress_index, | |
| pn_delivery_remote_state(pnd), | |
| qd_delivery_read_remote_state(pnd)); | |
| } else { | |
| //reject | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "[C%"PRIu64"][L%"PRIu64"] Message rejected due to policy violation on target. User:%s", | |
| conn->connection_id, | |
| qd_link_link_id(link), | |
| conn->user_id); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| _reject_delivery(pnd, QD_AMQP_COND_UNAUTHORIZED_ACCESS, "policy violation on target"); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| qd_iterator_free(addr_iter); | |
| qd_bitmask_free(link_exclusions); | |
| return next_delivery; | |
| } | |
| } | |
| } else { | |
| // | |
| // This is a targeted link, not anonymous. | |
| // | |
| // | |
| // Look in a series of locations for the terminus address, starting | |
| // with the qdr_link (in case this is an auto-link with separate | |
| // internal and external addresses). | |
| // | |
| const char *term_addr = qdr_link_internal_address(rlink); | |
| if (!term_addr) { | |
| term_addr = pn_terminus_get_address(qd_link_remote_target(link)); | |
| if (!term_addr) | |
| term_addr = pn_terminus_get_address(qd_link_source(link)); | |
| } | |
| if (term_addr) { | |
| int tenant_space_length; | |
| const char *tenant_space = _get_tenant_space(conn, &tenant_space_length); | |
| if (tenant_space) { | |
| qd_iterator_t *aiter = qd_iterator_string(term_addr, ITER_VIEW_ADDRESS_WITH_SPACE); | |
| qd_iterator_annotate_space(aiter, tenant_space, tenant_space_length); | |
| char *iter_str = (char *) qd_iterator_copy(aiter); | |
| qd_message_set_to_override_annotation(msg, iter_str); | |
| free(iter_str); | |
| qd_iterator_free(aiter); | |
| } else | |
| qd_message_set_to_override_annotation(msg, term_addr); | |
| int phase = qdr_link_phase(rlink); | |
| if (phase != 0) | |
| qd_message_set_phase_annotation(msg, phase); | |
| } | |
| log_link_message(conn, pn_link, msg); | |
| delivery = qdr_link_deliver(rlink, msg, ingress_iter, pn_delivery_settled(pnd), link_exclusions, ingress_index, | |
| pn_delivery_remote_state(pnd), | |
| qd_delivery_read_remote_state(pnd)); | |
| } | |
| // | |
| // End of new delivery processing | |
| // | |
| if (delivery) { | |
| qd_link_set_incoming_msg(link, (qd_message_t*) 0); // msg no longer exclusive to qd_link | |
| qdr_node_connect_deliveries(link, delivery, pnd); | |
| qdr_delivery_decref(router->router_core, delivery, "release protection of return from deliver"); | |
| } else { | |
| // | |
| // If there is no delivery, the message is now and will always be unroutable because there is no address. | |
| // | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "[C%"PRIu64"][L%"PRIu64"] Message rejected - no address present", | |
| conn->connection_id, | |
| qd_link_link_id(link)); | |
| qd_bitmask_free(link_exclusions); | |
| qd_message_set_discard(msg, true); | |
| pn_link_flow(pn_link, 1); | |
| _reject_delivery(pnd, QD_AMQP_COND_PRECONDITION_FAILED, "Routing failure: no address present"); | |
| if (receive_complete) { | |
| pn_delivery_settle(pnd); | |
| qd_message_free(msg); | |
| } | |
| } | |
| return next_delivery; | |
| } | |
| /** | |
| * Deferred callback for inbound delivery handler | |
| */ | |
| static void deferred_AMQP_rx_handler(void *context, bool discard) | |
| { | |
| qd_link_t_sp *safe_qdl = (qd_link_t_sp*) context; | |
| if (!discard) { | |
| qd_link_t *qdl = safe_deref_qd_link_t(*safe_qdl); | |
| if (!!qdl) { | |
| assert(qd_link_direction(qdl) == QD_INCOMING); | |
| qd_router_t *qdr = (qd_router_t*) qd_link_get_node_context(qdl); | |
| assert(qdr != 0); | |
| while (true) { | |
| if (!AMQP_rx_handler(qdr, qdl)) | |
| break; | |
| } | |
| } | |
| } | |
| free(safe_qdl); | |
| } | |
| /** | |
| * Delivery Disposition Handler | |
| */ | |
| static void AMQP_disposition_handler(void* context, qd_link_t *link, pn_delivery_t *pnd) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qdr_delivery_t *delivery = qdr_node_delivery_qdr_from_pn(pnd); | |
| // | |
| // It's important to not do any processing without a qdr_delivery. | |
| if (!delivery) | |
| return; | |
| uint64_t dstate = pn_delivery_remote_state(pnd); | |
| bool settled = pn_delivery_settled(pnd); | |
| // | |
| // When pre-settled multi-frame deliveries arrive, it's possible for the | |
| // settlement to register before the whole message arrives. Such premature | |
| // settlement indications must be ignored. | |
| // | |
| if (settled && !qdr_delivery_receive_complete(delivery)) | |
| settled = false; | |
| if (dstate || settled) { | |
| // | |
| // Update the disposition of the delivery | |
| // | |
| qdr_delivery_remote_state_updated(router->router_core, delivery, | |
| dstate, | |
| settled, | |
| qd_delivery_read_remote_state(pnd), | |
| false); | |
| // | |
| // If settled, close out the delivery | |
| // | |
| if (settled) { | |
| qdr_node_disconnect_deliveries(router->router_core, link, delivery, pnd); | |
| pn_delivery_settle(pnd); | |
| } | |
| } | |
| } | |
| /** | |
| * New Incoming Link Handler | |
| */ | |
| static int AMQP_incoming_link_handler(void* context, qd_link_t *link) | |
| { | |
| qd_connection_t *conn = qd_link_connection(link); | |
| uint64_t link_id; | |
| // The connection that this link belongs to is gone. Perhaps an AMQP close came in. | |
| // This link handler should not continue since there is no connection. | |
| if (conn == 0) | |
| return 0; | |
| qdr_connection_t *qdr_conn = (qdr_connection_t*) qd_connection_get_context(conn); | |
| char *terminus_addr = (char*)pn_terminus_get_address(pn_link_remote_target((pn_link_t *)qd_link_pn(link))); | |
| qdr_link_t *qdr_link = qdr_link_first_attach(qdr_conn, QD_INCOMING, | |
| qdr_terminus(qd_link_remote_source(link)), | |
| qdr_terminus(qd_link_remote_target(link)), | |
| pn_link_name(qd_link_pn(link)), | |
| terminus_addr, | |
| false, | |
| 0, | |
| &link_id); | |
| qd_link_set_link_id(link, link_id); | |
| qdr_link_set_context(qdr_link, link); | |
| qd_link_set_context(link, qdr_link); | |
| return 0; | |
| } | |
| /** | |
| * New Outgoing Link Handler | |
| */ | |
| static int AMQP_outgoing_link_handler(void* context, qd_link_t *link) | |
| { | |
| qd_connection_t *conn = qd_link_connection(link); | |
| uint64_t link_id; | |
| // The connection that this link belongs to is gone. Perhaps an AMQP close came in. | |
| // This link handler should not continue since there is no connection. | |
| if (conn == 0) | |
| return 0; | |
| qdr_connection_t *qdr_conn = (qdr_connection_t*) qd_connection_get_context(conn); | |
| char *terminus_addr = (char*)pn_terminus_get_address(pn_link_remote_source((pn_link_t *)qd_link_pn(link))); | |
| qdr_link_t *qdr_link = qdr_link_first_attach(qdr_conn, QD_OUTGOING, | |
| qdr_terminus(qd_link_remote_source(link)), | |
| qdr_terminus(qd_link_remote_target(link)), | |
| pn_link_name(qd_link_pn(link)), | |
| terminus_addr, | |
| false, | |
| 0, | |
| &link_id); | |
| qd_link_set_link_id(link, link_id); | |
| qdr_link_set_context(qdr_link, link); | |
| qd_link_set_context(link, qdr_link); | |
| return 0; | |
| } | |
| /** | |
| * Handler for remote opening of links that we initiated. | |
| */ | |
| static int AMQP_link_attach_handler(void* context, qd_link_t *link) | |
| { | |
| qdr_link_t *qlink = (qdr_link_t*) qd_link_get_context(link); | |
| qdr_link_second_attach(qlink, | |
| qdr_terminus(qd_link_remote_source(link)), | |
| qdr_terminus(qd_link_remote_target(link))); | |
| return 0; | |
| } | |
| /** | |
| * Handler for flow events on links. Flow updates include session window | |
| * state, which needs to be checked for unblocking Q3. | |
| */ | |
| static int AMQP_link_flow_handler(void* context, qd_link_t *link) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| pn_link_t *pnlink = qd_link_pn(link); | |
| qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link); | |
| if (rlink) { | |
| qdr_link_flow(router->router_core, rlink, pn_link_remote_credit(pnlink), pn_link_get_drain(pnlink)); | |
| } | |
| // check if Q3 can be unblocked | |
| pn_session_t *pn_ssn = pn_link_session(pnlink); | |
| if (pn_ssn) { | |
| qd_session_t *qd_ssn = qd_session_from_pn(pn_ssn); | |
| if (qd_ssn && qd_session_is_q3_blocked(qd_ssn)) { | |
| // Q3 blocked - have we drained enough outgoing bytes? | |
| const size_t q3_lower = BUFFER_SIZE * QD_QLIMIT_Q3_LOWER; | |
| if (pn_session_outgoing_bytes(pn_ssn) < q3_lower) { | |
| // yes. We must now unblock all links that have been blocked by Q3 | |
| qd_link_list_t *blinks = qd_session_q3_blocked_links(qd_ssn); | |
| qd_link_t *blink = DEQ_HEAD(*blinks); | |
| while (blink) { | |
| qd_link_q3_unblock(blink); // removes from blinks list! | |
| if (blink != link) { // already flowed this link | |
| rlink = (qdr_link_t *) qd_link_get_context(blink); | |
| if (rlink) { | |
| pnlink = qd_link_pn(blink); | |
| // signalling flow to the core causes the link to be re-activated | |
| qdr_link_flow(router->router_core, rlink, pn_link_remote_credit(pnlink), pn_link_get_drain(pnlink)); | |
| } | |
| } | |
| blink = DEQ_HEAD(*blinks); | |
| } | |
| } | |
| } | |
| } | |
| return 0; | |
| } | |
| /** | |
| * Link Detached Handler | |
| */ | |
| static int AMQP_link_detach_handler(void* context, qd_link_t *link, qd_detach_type_t dt) | |
| { | |
| if (!link) | |
| return 0; | |
| pn_link_t *pn_link = qd_link_pn(link); | |
| if (!pn_link) | |
| return 0; | |
| // DISPATCH-1085: If link is in the middle of receiving a message it is | |
| // possible that the message is actually complete but the remaining message | |
| // data is still in proton's buffers. (e.g. a large message is sent then | |
| // the sender immediately detaches) Force a call to the rx_handler for the | |
| // link in order to pull the buffered data into the message. | |
| if (pn_link_is_receiver(pn_link)) { | |
| pn_delivery_t *pnd = pn_link_current(pn_link); | |
| if (pnd) { | |
| qd_message_t *msg = qd_get_message_context(pnd); | |
| if (msg) { | |
| if (!qd_message_receive_complete(msg)) { | |
| qd_link_set_q2_limit_unbounded(link, true); | |
| // since this thread owns link we can call the | |
| // rx_hander directly rather than schedule it via | |
| // the unblock handler: | |
| qd_message_clear_q2_unblocked_handler(msg); | |
| qd_message_Q2_holdoff_disable(msg); | |
| while (AMQP_rx_handler((qd_router_t*) context, link)) | |
| ; | |
| } | |
| } | |
| } | |
| } | |
| qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link); | |
| pn_condition_t *cond = qd_link_pn(link) ? pn_link_remote_condition(qd_link_pn(link)) : 0; | |
| if (rlink) { | |
| // | |
| // This is the last event for this link that we will send into the core. Remove the | |
| // core linkage. Note that the core->qd linkage is still in place. | |
| // | |
| qd_link_set_context(link, 0); | |
| // | |
| // If the link was lost (due to connection drop), or the linkage from the core | |
| // object is already gone, finish disconnecting the linkage and free the qd_link | |
| // because the core will silently free its own resources. | |
| // | |
| if (dt == QD_LOST || qdr_link_get_context(rlink) == 0) { | |
| qdr_link_set_context(rlink, 0); | |
| qd_link_free(link); | |
| } | |
| qdr_error_t *error = qdr_error_from_pn(cond); | |
| qdr_link_detach(rlink, dt, error); | |
| } | |
| return 0; | |
| } | |
| static void bind_connection_context(qdr_connection_t *qdrc, void* token) | |
| { | |
| qd_connection_t *conn = (qd_connection_t*) token; | |
| qd_connection_set_context(conn, qdrc); | |
| qdr_connection_set_context(qdrc, conn); | |
| } | |
| static void save_original_and_current_conn_info(qd_connection_t *conn) | |
| { | |
| // The failover list is present but it is empty. We will wipe the old backup information from the failover list. | |
| // The only items we want to keep in this list is the original connection information (from the config file) | |
| // and the current connection information. | |
| if (conn->connector && DEQ_SIZE(conn->connector->conn_info_list) > 1) { | |
| // Here we are simply removing all other failover information except the original connection information and the one we used to make a successful connection. | |
| int i = 1; | |
| qd_failover_item_t *item = DEQ_HEAD(conn->connector->conn_info_list); | |
| qd_failover_item_t *next_item = 0; | |
| bool match_found = false; | |
| int dec_conn_index=0; | |
| while(item) { | |
| //The first item on this list is always the original connector, so we want to keep that item in place | |
| // We have to delete items in the list that were left over from the previous failover list from the previous connection | |
| // because the new connection might have its own failover list. | |
| if (i != conn->connector->conn_index) { | |
| if (item != DEQ_HEAD(conn->connector->conn_info_list)) { | |
| next_item = DEQ_NEXT(item); | |
| free(item->scheme); | |
| free(item->host); | |
| free(item->port); | |
| free(item->hostname); | |
| free(item->host_port); | |
| DEQ_REMOVE(conn->connector->conn_info_list, item); | |
| free(item); | |
| item = next_item; | |
| // We are removing an item from the list before the conn_index match was found. We need to | |
| // decrement the conn_index | |
| if (!match_found) | |
| dec_conn_index += 1; | |
| } | |
| else { | |
| item = DEQ_NEXT(item); | |
| } | |
| } | |
| else { | |
| match_found = true; | |
| item = DEQ_NEXT(item); | |
| } | |
| i += 1; | |
| } | |
| conn->connector->conn_index -= dec_conn_index; | |
| } | |
| } | |
| static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool inbound) | |
| { | |
| qdr_connection_role_t role = 0; | |
| int cost = 1; | |
| int link_capacity = 1; | |
| const char *name = 0; | |
| bool multi_tenant = false; | |
| bool streaming_links = false; | |
| const char *vhost = 0; | |
| char rversion[128]; | |
| uint64_t connection_id = qd_connection_connection_id(conn); | |
| pn_connection_t *pn_conn = qd_connection_pn(conn); | |
| pn_transport_t *tport = 0; | |
| pn_sasl_t *sasl = 0; | |
| pn_ssl_t *ssl = 0; | |
| const char *mech = 0; | |
| const char *user = 0; | |
| const char *container = conn->pn_conn ? pn_connection_remote_container(conn->pn_conn) : 0; | |
| rversion[0] = 0; | |
| conn->strip_annotations_in = false; | |
| conn->strip_annotations_out = false; | |
| if (conn->pn_conn) { | |
| tport = pn_connection_transport(conn->pn_conn); | |
| ssl = conn->ssl; | |
| } | |
| if (tport) { | |
| sasl = pn_sasl(tport); | |
| if(conn->user_id) | |
| user = conn->user_id; | |
| else | |
| user = pn_transport_get_user(tport); | |
| } | |
| if (sasl) | |
| mech = pn_sasl_get_mech(sasl); | |
| const char *host = 0; | |
| char host_local[255]; | |
| const qd_server_config_t *config; | |
| if (qd_connection_connector(conn)) { | |
| config = qd_connector_config(qd_connection_connector(conn)); | |
| snprintf(host_local, 254, "%s", config->host_port); | |
| host = &host_local[0]; | |
| } | |
| else | |
| host = qd_connection_name(conn); | |
| qd_router_connection_get_config(conn, &role, &cost, &name, &multi_tenant, | |
| &conn->strip_annotations_in, &conn->strip_annotations_out, &link_capacity); | |
| // check offered capabilities for streaming link support | |
| // | |
| pn_data_t *ocaps = pn_connection_remote_offered_capabilities(pn_conn); | |
| if (ocaps) { | |
| size_t sl_len = strlen(QD_CAPABILITY_STREAMING_LINKS); | |
| pn_data_rewind(ocaps); | |
| if (pn_data_next(ocaps)) { | |
| if (pn_data_type(ocaps) == PN_ARRAY) { | |
| pn_data_enter(ocaps); | |
| pn_data_next(ocaps); | |
| } | |
| do { | |
| if (pn_data_type(ocaps) == PN_SYMBOL) { | |
| pn_bytes_t s = pn_data_get_symbol(ocaps); | |
| streaming_links = (s.size == sl_len | |
| && strncmp(s.start, QD_CAPABILITY_STREAMING_LINKS, sl_len) == 0); | |
| } | |
| } while (pn_data_next(ocaps) && !streaming_links); | |
| } | |
| } | |
| // if connection properties are present parse out any important data | |
| // | |
| pn_data_t *props = pn_conn ? pn_connection_remote_properties(pn_conn) : 0; | |
| if (props) { | |
| const bool is_router = (role == QDR_ROLE_INTER_ROUTER || role == QDR_ROLE_EDGE_CONNECTION); | |
| pn_data_rewind(props); | |
| if (pn_data_next(props) && pn_data_type(props) == PN_MAP) { | |
| const size_t num_items = pn_data_get_map(props); | |
| int props_found = 0; // once all props found exit loop | |
| pn_data_enter(props); | |
| for (int i = 0; i < num_items / 2 && props_found < 4; ++i) { | |
| if (!pn_data_next(props)) break; | |
| if (pn_data_type(props) != PN_SYMBOL) break; // invalid properties map | |
| pn_bytes_t key = pn_data_get_symbol(props); | |
| if (key.size == strlen(QD_CONNECTION_PROPERTY_COST_KEY) && | |
| strncmp(key.start, QD_CONNECTION_PROPERTY_COST_KEY, key.size) == 0) { | |
| props_found += 1; | |
| if (!pn_data_next(props)) break; | |
| if (is_router) { | |
| if (pn_data_type(props) == PN_INT) { | |
| const int remote_cost = (int) pn_data_get_int(props); | |
| if (remote_cost > cost) | |
| cost = remote_cost; | |
| } | |
| } | |
| } else if (key.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY) && | |
| strncmp(key.start, QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY, key.size) == 0) { | |
| props_found += 1; | |
| if (!pn_data_next(props)) break; | |
| parse_failover_property_list(router, conn, props); | |
| } else if (key.size == strlen(QD_CONNECTION_PROPERTY_VERSION_KEY) | |
| && strncmp(key.start, QD_CONNECTION_PROPERTY_VERSION_KEY, key.size) == 0) { | |
| props_found += 1; | |
| if (!pn_data_next(props)) break; | |
| if (is_router) { | |
| pn_bytes_t vdata = pn_data_get_string(props); | |
| size_t vlen = MIN(sizeof(rversion) - 1, vdata.size); | |
| strncpy(rversion, vdata.start, vlen); | |
| rversion[vlen] = 0; | |
| } | |
| } else if ((key.size == strlen(QD_CONNECTION_PROPERTY_ANNOTATIONS_VERSION_KEY) | |
| && strncmp(key.start, QD_CONNECTION_PROPERTY_ANNOTATIONS_VERSION_KEY, key.size) == 0)) { | |
| props_found += 1; | |
| if (!pn_data_next(props)) break; | |
| if (is_router && pn_data_type(props) == PN_INT) { | |
| const int annos_version = (int) pn_data_get_int(props); | |
| qd_log(router->log_source, QD_LOG_DEBUG, | |
| "Remote router annotations version: %d", annos_version); | |
| } | |
| } else { | |
| // skip this key | |
| if (!pn_data_next(props)) break; | |
| } | |
| } | |
| } | |
| } | |
| if (multi_tenant) | |
| vhost = (conn->policy_settings && conn->policy_settings->vhost_name) ? | |
| conn->policy_settings->vhost_name : | |
| pn_connection_remote_hostname(pn_conn); | |
| char proto[50]; | |
| memset(proto, 0, 50); | |
| char cipher[50]; | |
| memset(cipher, 0, 50); | |
| int ssl_ssf = 0; | |
| bool is_ssl = false; | |
| if (ssl) { | |
| pn_ssl_get_protocol_name(ssl, proto, 50); | |
| pn_ssl_get_cipher_name(ssl, cipher, 50); | |
| ssl_ssf = pn_ssl_get_ssf(ssl); | |
| is_ssl = true; | |
| } | |
| bool encrypted = tport && pn_transport_is_encrypted(tport); | |
| bool authenticated = tport && pn_transport_is_authenticated(tport); | |
| qdr_connection_info_t *connection_info = qdr_connection_info(encrypted, | |
| authenticated, | |
| conn->opened, | |
| (char*) mech, | |
| conn->connector ? QD_OUTGOING : QD_INCOMING, | |
| host, | |
| proto, | |
| cipher, | |
| (char*) user, | |
| container, | |
| props, | |
| ssl_ssf, | |
| is_ssl, | |
| rversion, | |
| streaming_links); | |
| qdr_connection_opened(router->router_core, | |
| amqp_direct_adaptor, | |
| inbound, | |
| role, | |
| cost, | |
| connection_id, | |
| name, | |
| pn_connection_remote_container(pn_conn), | |
| conn->strip_annotations_in, | |
| conn->strip_annotations_out, | |
| link_capacity, | |
| vhost, | |
| !!conn->policy_settings ? &conn->policy_settings->spec : 0, | |
| connection_info, | |
| bind_connection_context, | |
| conn); | |
| if (conn->connector) { | |
| char conn_msg[300]; | |
| qd_format_string(conn_msg, 300, "[C%"PRIu64"] Connection Opened: dir=%s host=%s vhost=%s encrypted=%s" | |
| " auth=%s user=%s container_id=%s", | |
| connection_id, inbound ? "in" : "out", host, vhost ? vhost : "", encrypted ? proto : "no", | |
| authenticated ? mech : "no", (char*) user, container); | |
| sys_mutex_lock(conn->connector->lock); | |
| strcpy(conn->connector->conn_msg, conn_msg); | |
| sys_mutex_unlock(conn->connector->lock); | |
| } | |
| } | |
| // We are attempting to find a connection property called failover-server-list which is a list of failover host names and ports.. | |
| // failover-server-list looks something like this | |
| // :"failover-server-list"=[{:"network-host"="some-host", :port="35000"}, {:"network-host"="localhost", :port="25000"}] | |
| // There are three cases here - | |
| // 1. The failover-server-list is present but the content of the list is empty in which case we scrub the failover list except we keep the original connector information and current connection information. | |
| // 2. If the failover list contains one or more maps that contain failover connection information, that information will be appended to the list which already contains the original connection information | |
| // and the current connection information. Any other failover information left over from the previous connection is deleted | |
| // 3. If the failover-server-list is not present at all in the connection properties, the failover list we maintain is untouched. | |
| // | |
| // props should be pointing at the value that corresponds to the QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY | |
| // returns true if failover list properly parsed. | |
| // | |
| static bool parse_failover_property_list(qd_router_t *router, qd_connection_t *conn, pn_data_t *props) | |
| { | |
| bool found_failover = false; | |
| if (pn_data_type(props) != PN_LIST) | |
| return false; | |
| size_t list_num_items = pn_data_get_list(props); | |
| if (list_num_items > 0) { | |
| save_original_and_current_conn_info(conn); | |
| pn_data_enter(props); // enter list | |
| for (int i=0; i < list_num_items; i++) { | |
| pn_data_next(props);// this is the first element of the list, a map. | |
| if (props && pn_data_type(props) == PN_MAP) { | |
| size_t map_num_items = pn_data_get_map(props); | |
| pn_data_enter(props); | |
| qd_failover_item_t *item = NEW(qd_failover_item_t); | |
| ZERO(item); | |
| // We have found a map with the connection information. Step thru the map contents and create qd_failover_item_t | |
| for (int j=0; j < map_num_items/2; j++) { | |
| pn_data_next(props); | |
| if (pn_data_type(props) == PN_SYMBOL) { | |
| pn_bytes_t sym = pn_data_get_symbol(props); | |
| if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY) && | |
| strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY) == 0) { | |
| pn_data_next(props); | |
| if (pn_data_type(props) == PN_STRING) { | |
| item->host = strdup(pn_data_get_string(props).start); | |
| } | |
| } | |
| else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY) && | |
| strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_PORT_KEY) == 0) { | |
| pn_data_next(props); | |
| item->port = qdpn_data_as_string(props); | |
| } | |
| else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY) && | |
| strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_SCHEME_KEY) == 0) { | |
| pn_data_next(props); | |
| if (pn_data_type(props) == PN_STRING) { | |
| item->scheme = strdup(pn_data_get_string(props).start); | |
| } | |
| } | |
| else if (sym.size == strlen(QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY) && | |
| strcmp(sym.start, QD_CONNECTION_PROPERTY_FAILOVER_HOSTNAME_KEY) == 0) { | |
| pn_data_next(props); | |
| if (pn_data_type(props) == PN_STRING) { | |
| item->hostname = strdup(pn_data_get_string(props).start); | |
| } | |
| } | |
| } | |
| } | |
| int host_length = strlen(item->host); | |
| // | |
| // We will not even bother inserting the item if there is no host available. | |
| // | |
| if (host_length != 0) { | |
| if (item->scheme == 0) | |
| item->scheme = strdup("amqp"); | |
| if (item->port == 0) | |
| item->port = strdup("5672"); | |
| int hplen = strlen(item->host) + strlen(item->port) + 2; | |
| item->host_port = malloc(hplen); | |
| snprintf(item->host_port, hplen, "%s:%s", item->host, item->port); | |
| // | |
| // Iterate through failover list items and sets insert_tail to true | |
| // when list has just original connector's host and port or when new | |
| // reported host and port is not yet part of the current list. | |
| // | |
| bool insert_tail = false; | |
| if ( DEQ_SIZE(conn->connector->conn_info_list) == 1 ) { | |
| insert_tail = true; | |
| } else { | |
| qd_failover_item_t *conn_item = DEQ_HEAD(conn->connector->conn_info_list); | |
| insert_tail = true; | |
| while ( conn_item ) { | |
| if ( !strcmp(conn_item->host_port, item->host_port) ) { | |
| insert_tail = false; | |
| break; | |
| } | |
| conn_item = DEQ_NEXT(conn_item); | |
| } | |
| } | |
| // Only inserts if not yet part of failover list | |
| if ( insert_tail ) { | |
| DEQ_INSERT_TAIL(conn->connector->conn_info_list, item); | |
| qd_log(router->log_source, QD_LOG_DEBUG, "Added %s as backup host", item->host_port); | |
| found_failover = true; | |
| } | |
| else { | |
| free(item->scheme); | |
| free(item->host); | |
| free(item->port); | |
| free(item->hostname); | |
| free(item->host_port); | |
| free(item); | |
| } | |
| } | |
| else { | |
| free(item->scheme); | |
| free(item->host); | |
| free(item->port); | |
| free(item->hostname); | |
| free(item->host_port); | |
| free(item); | |
| } | |
| } | |
| pn_data_exit(props); | |
| } | |
| } // list_num_items > 0 | |
| else { | |
| save_original_and_current_conn_info(conn); | |
| } | |
| return found_failover; | |
| } | |
| static int AMQP_inbound_opened_handler(void *type_context, qd_connection_t *conn, void *context) | |
| { | |
| qd_router_t *router = (qd_router_t*) type_context; | |
| AMQP_opened_handler(router, conn, true); | |
| return 0; | |
| } | |
| static int AMQP_outbound_opened_handler(void *type_context, qd_connection_t *conn, void *context) | |
| { | |
| qd_router_t *router = (qd_router_t*) type_context; | |
| AMQP_opened_handler(router, conn, false); | |
| return 0; | |
| } | |
| static int AMQP_closed_handler(void *type_context, qd_connection_t *conn, void *context) | |
| { | |
| qdr_connection_t *qdrc = (qdr_connection_t*) qd_connection_get_context(conn); | |
| qd_router_t *router = (qd_router_t*) type_context; | |
| if (qdrc) { | |
| sys_mutex_lock(qd_server_get_activation_lock(router->qd->server)); | |
| qdr_connection_set_context(qdrc, 0); | |
| sys_mutex_unlock(qd_server_get_activation_lock(router->qd->server)); | |
| qdr_connection_closed(qdrc); | |
| qd_connection_set_context(conn, 0); | |
| } | |
| return 0; | |
| } | |
| static void qd_router_timer_handler(void *context) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| // | |
| // Periodic processing. | |
| // | |
| qd_pyrouter_tick(router); | |
| // This sends a tick into the core and this happens every second. | |
| qdr_process_tick(router->router_core); | |
| qd_timer_schedule(router->timer, 1000); | |
| } | |
| static qd_node_type_t router_node = {"router", 0, 0, | |
| AMQP_rx_handler, | |
| AMQP_disposition_handler, | |
| AMQP_incoming_link_handler, | |
| AMQP_outgoing_link_handler, | |
| AMQP_writable_conn_handler, | |
| AMQP_link_detach_handler, | |
| AMQP_link_attach_handler, | |
| qd_link_abandoned_deliveries_handler, | |
| AMQP_link_flow_handler, | |
| 0, // node_created_handler | |
| 0, // node_destroyed_handler | |
| AMQP_inbound_opened_handler, | |
| AMQP_outbound_opened_handler, | |
| AMQP_closed_handler}; | |
| // not api, but needed by unit tests | |
| void qd_router_id_initialize(const char *area, const char *id) | |
| { | |
| size_t dplen = 2 + strlen(area) + strlen(id); | |
| node_id = (char*) qd_malloc(dplen); | |
| strcpy(node_id, area); | |
| strcat(node_id, "/"); | |
| strcat(node_id, id); | |
| // Node ID as an AMQP encoded str value. Used when composing trace list | |
| // and ingress message annotations into the outgoing message | |
| const uint32_t id_len = strlen(node_id); | |
| const uint32_t extra = 5; // 5 octets = max AMQP STRx header length | |
| encoded_node_id = (uint8_t*) qd_malloc(id_len + extra + 1); // 1 = string terminator | |
| const int hdr_len = qd_compose_str_header(encoded_node_id, id_len); | |
| assert(hdr_len <= extra); | |
| strcpy((char*) &encoded_node_id[hdr_len], node_id); | |
| encoded_node_id_len = hdr_len + id_len; | |
| } | |
| // not api, but needed by unit tests | |
| void qd_router_id_finalize(void) | |
| { | |
| free(node_id); | |
| node_id = 0; | |
| free(encoded_node_id); | |
| encoded_node_id = 0; | |
| encoded_node_id_len = 0; | |
| } | |
| qd_router_t *qd_router(qd_dispatch_t *qd, qd_router_mode_t mode, const char *area, const char *id) | |
| { | |
| qd_container_register_node_type(qd, &router_node); | |
| qd_router_id_initialize(area, id); | |
| qd_router_t *router = NEW(qd_router_t); | |
| ZERO(router); | |
| router_node.type_context = router; | |
| qd->router = router; | |
| router->qd = qd; | |
| router->router_core = 0; | |
| router->log_source = qd_log_source("ROUTER"); | |
| router->router_mode = mode; | |
| router->router_area = area; | |
| router->router_id = id; | |
| router->node = qd_container_set_default_node_type(qd, &router_node, (void*) router, QD_DIST_BOTH); | |
| router->lock = sys_mutex(); | |
| router->timer = qd_timer(qd, qd_router_timer_handler, (void*) router); | |
| sys_atomic_init(&global_delivery_id, 1); | |
| // | |
| // Inform the field iterator module of this router's mode, id, and area. The field iterator | |
| // uses this to offload some of the address-processing load from the router. | |
| // | |
| qd_iterator_set_address(mode == QD_ROUTER_MODE_EDGE, area, id); | |
| switch (router->router_mode) { | |
| case QD_ROUTER_MODE_STANDALONE: qd_log(router->log_source, QD_LOG_INFO, "Router started in Standalone mode"); break; | |
| case QD_ROUTER_MODE_INTERIOR: qd_log(router->log_source, QD_LOG_INFO, "Router started in Interior mode, area=%s id=%s", area, id); break; | |
| case QD_ROUTER_MODE_EDGE: qd_log(router->log_source, QD_LOG_INFO, "Router started in Edge mode"); break; | |
| case QD_ROUTER_MODE_ENDPOINT: qd_log(router->log_source, QD_LOG_INFO, "Router started in Endpoint mode"); break; | |
| } | |
| qd_log(router->log_source, QD_LOG_INFO, "Version: %s", QPID_DISPATCH_VERSION); | |
| return router; | |
| } | |
| static void CORE_connection_activate(void *context, qdr_connection_t *conn) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| // | |
| // IMPORTANT: This is the only core callback that is invoked on the core | |
| // thread itself. It must not take locks that could deadlock the core. | |
| // | |
| sys_mutex_lock(qd_server_get_activation_lock(router->qd->server)); | |
| qd_server_activate((qd_connection_t*) qdr_connection_get_context(conn)); | |
| sys_mutex_unlock(qd_server_get_activation_lock(router->qd->server)); | |
| } | |
| static void CORE_link_first_attach(void *context, | |
| qdr_connection_t *conn, | |
| qdr_link_t *link, | |
| qdr_terminus_t *source, | |
| qdr_terminus_t *target, | |
| qd_session_class_t ssn_class) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qd_connection_t *qconn = (qd_connection_t*) qdr_connection_get_context(conn); | |
| if (!qconn) return; /* Connection is already closed */ | |
| // | |
| // Create a new link to be attached | |
| // | |
| qd_link_t *qlink = qd_link(router->node, qconn, qdr_link_direction(link), qdr_link_name(link), ssn_class); | |
| // | |
| // Copy the source and target termini to the link | |
| // | |
| qdr_terminus_copy(source, qd_link_source(qlink)); | |
| qdr_terminus_copy(target, qd_link_target(qlink)); | |
| // | |
| // Associate the qd_link and the qdr_link to each other | |
| // | |
| qdr_link_set_context(link, qlink); | |
| qd_link_set_context(qlink, link); | |
| // | |
| // Open (attach) the link | |
| // | |
| pn_link_open(qd_link_pn(qlink)); | |
| // | |
| // Mark the link as stalled and waiting for initial credit. | |
| // | |
| if (qdr_link_direction(link) == QD_OUTGOING) | |
| qdr_link_stalled_outbound(link); | |
| } | |
| static void CORE_link_second_attach(void *context, qdr_link_t *link, qdr_terminus_t *source, qdr_terminus_t *target) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *pn_link = qd_link_pn(qlink); | |
| if (!pn_link) | |
| return; | |
| qdr_terminus_copy(source, qd_link_source(qlink)); | |
| qdr_terminus_copy(target, qd_link_target(qlink)); | |
| // | |
| // Open (attach) the link | |
| // | |
| pn_link_open(pn_link); | |
| // | |
| // Mark the link as stalled and waiting for initial credit. | |
| // | |
| if (qdr_link_direction(link) == QD_OUTGOING) | |
| qdr_link_stalled_outbound(link); | |
| } | |
| static void CORE_conn_trace(void *context, qdr_connection_t *qdr_conn, bool trace) | |
| { | |
| qd_connection_t *qconn = (qd_connection_t*) qdr_connection_get_context(qdr_conn); | |
| if (!qconn) | |
| return; | |
| pn_transport_t *tport = pn_connection_transport(qconn->pn_conn); | |
| if (!tport) | |
| return; | |
| if (trace) { | |
| pn_transport_trace(tport, PN_TRACE_FRM); | |
| pn_transport_set_tracer(tport, connection_transport_tracer); | |
| } | |
| else { | |
| pn_transport_trace(tport, PN_TRACE_OFF); | |
| } | |
| } | |
| static void CORE_close_connection(void *context, qdr_connection_t *qdr_conn, qdr_error_t *error) | |
| { | |
| if (qdr_conn) { | |
| qd_connection_t *qd_conn = qdr_connection_get_context(qdr_conn); | |
| if (qd_conn) { | |
| pn_connection_t *pn_conn = qd_connection_pn(qd_conn); | |
| if (pn_conn) { | |
| // | |
| // Go down to the transport and close the head and tail. This will | |
| // drop the socket to the peer without providing any error indication. | |
| // Due to issues in Proton that cause different behaviors in different | |
| // bindings depending on whether there is a connection:forced error, | |
| // this has been deemed the best way to force the peer to reconnect. | |
| // | |
| pn_transport_t *tport = pn_connection_transport(pn_conn); | |
| pn_transport_close_head(tport); | |
| pn_transport_close_tail(tport); | |
| } | |
| } | |
| } | |
| } | |
| static void CORE_link_detach(void *context, qdr_link_t *link, qdr_error_t *error, bool first, bool close) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *pn_link = qd_link_pn(qlink); | |
| if (!pn_link) | |
| return; | |
| if (error) { | |
| pn_condition_t *cond = pn_link_condition(pn_link); | |
| qdr_error_copy(error, cond); | |
| } | |
| // | |
| // If the link is only half open, then this DETACH constitutes the rejection of | |
| // an incoming ATTACH. We must nullify the source and target in order to be | |
| // compliant with the AMQP specification. This is because Proton will generate | |
| // the missing ATTACH before the DETACH and will include spurious terminus data | |
| // if we don't nullify it here. | |
| // | |
| if (pn_link_state(pn_link) & PN_LOCAL_UNINIT) { | |
| if (pn_link_is_receiver(pn_link)) { | |
| pn_terminus_set_type(pn_link_target(pn_link), PN_UNSPECIFIED); | |
| pn_terminus_copy(pn_link_source(pn_link), pn_link_remote_source(pn_link)); | |
| } else { | |
| pn_terminus_set_type(pn_link_source(pn_link), PN_UNSPECIFIED); | |
| pn_terminus_copy(pn_link_target(pn_link), pn_link_remote_target(pn_link)); | |
| } | |
| } | |
| if (close) | |
| qd_link_close(qlink); | |
| else | |
| qd_link_detach(qlink); | |
| // | |
| // This is the last event for this link that we are going to send into Proton. | |
| // Remove the core->proton linkage. Note that the proton->core linkage may still | |
| // be intact and needed. | |
| // | |
| qdr_link_set_context(link, 0); | |
| // | |
| // If this is the second detach, free the qd_link | |
| // | |
| if (!first) { | |
| qd_link_free(qlink); | |
| } | |
| } | |
| static void CORE_link_flow(void *context, qdr_link_t *link, int credit) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (plink) | |
| pn_link_flow(plink, credit); | |
| } | |
| static void CORE_link_offer(void *context, qdr_link_t *link, int delivery_count) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (plink) | |
| pn_link_offered(plink, delivery_count); | |
| } | |
| static void CORE_link_drained(void *context, qdr_link_t *link) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (plink) { | |
| pn_link_drained(plink); | |
| qdr_link_set_drained(router->router_core, link); | |
| } | |
| } | |
| static void CORE_link_drain(void *context, qdr_link_t *link, bool mode) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (plink) { | |
| if (pn_link_is_receiver(plink)) | |
| pn_link_set_drain(plink, mode); | |
| } | |
| } | |
| static int CORE_link_push(void *context, qdr_link_t *link, int limit) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| if (!qlink) | |
| return 0; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (plink) { | |
| int link_credit = pn_link_credit(plink); | |
| if (link_credit > limit) | |
| link_credit = limit; | |
| if (link_credit > 0) | |
| // We will not bother calling qdr_link_process_deliveries if we have no credit. | |
| return qdr_link_process_deliveries(router->router_core, link, link_credit); | |
| } | |
| return 0; | |
| } | |
| static uint64_t CORE_link_deliver(void *context, qdr_link_t *link, qdr_delivery_t *dlv, bool settled) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| qd_connection_t *qconn = qd_link_connection(qlink); | |
| uint64_t update = 0; | |
| if (!qlink) | |
| return 0; | |
| pn_link_t *plink = qd_link_pn(qlink); | |
| if (!plink) | |
| return 0; | |
| // | |
| // If the remote send settle mode is set to 'settled' then settle the delivery on behalf of the receiver. | |
| // | |
| bool remote_snd_settled = qd_link_remote_snd_settle_mode(qlink) == PN_SND_SETTLED; | |
| pn_delivery_t *pdlv = 0; | |
| if (!qdr_delivery_tag_sent(dlv)) { | |
| const char *tag; | |
| int tag_length; | |
| uint64_t disposition = 0; | |
| qdr_delivery_tag(dlv, &tag, &tag_length); | |
| // Create a new proton delivery on link 'plink' | |
| pn_delivery(plink, pn_dtag(tag, tag_length)); | |
| pdlv = pn_link_current(plink); | |
| // handle any delivery-state on the transfer e.g. transactional-state | |
| qd_delivery_state_t *dstate = qdr_delivery_take_local_delivery_state(dlv, &disposition); | |
| if (disposition) { | |
| if (dstate) | |
| qd_delivery_write_local_state(pdlv, disposition, dstate); | |
| pn_delivery_update(pdlv, disposition); | |
| } | |
| qd_delivery_state_free(dstate); | |
| // | |
| // If the remote send settle mode is set to 'settled', we should settle the delivery on behalf of the receiver. | |
| // | |
| if (qdr_delivery_get_context(dlv) == 0) | |
| qdr_node_connect_deliveries(qlink, dlv, pdlv); | |
| qdr_delivery_set_tag_sent(dlv, true); | |
| } else { | |
| pdlv = qdr_node_delivery_pn_from_qdr(dlv); | |
| } | |
| if (!pdlv) | |
| return 0; | |
| bool q3_stalled = false; | |
| qd_message_t *msg_out = qdr_delivery_message(dlv); | |
| qd_message_send(msg_out, qlink, qdr_link_strip_annotations_out(link), &q3_stalled); | |
| if (q3_stalled) { | |
| qd_link_q3_block(qlink); | |
| qdr_link_stalled_outbound(link); | |
| } | |
| bool send_complete = qdr_delivery_send_complete(dlv); | |
| if (send_complete) { | |
| if (qd_message_aborted(msg_out)) { | |
| // Aborted messages must be settled locally | |
| // Settling does not produce any disposition to message sender. | |
| if (pdlv) { | |
| pn_link_advance(plink); | |
| qdr_node_disconnect_deliveries(router->router_core, qlink, dlv, pdlv); | |
| pn_delivery_settle(pdlv); | |
| } | |
| } else { | |
| if (!settled && remote_snd_settled) { | |
| // The caller must tell the core that the delivery has been | |
| // accepted and settled, since we are settling on behalf of the | |
| // receiver | |
| update = PN_ACCEPTED; // schedule the settle | |
| } | |
| pn_link_advance(plink); | |
| if (settled || remote_snd_settled) { | |
| if (pdlv) { | |
| qdr_node_disconnect_deliveries(router->router_core, qlink, dlv, pdlv); | |
| pn_delivery_settle(pdlv); | |
| } | |
| } | |
| } | |
| log_link_message(qconn, plink, msg_out); | |
| } | |
| return update; | |
| } | |
| static int CORE_link_get_credit(void *context, qdr_link_t *link) | |
| { | |
| qd_link_t *qlink = (qd_link_t*) qdr_link_get_context(link); | |
| pn_link_t *plink = !!qlink ? qd_link_pn(qlink) : 0; | |
| if (!plink) | |
| return 0; | |
| return pn_link_remote_credit(plink); | |
| } | |
| static void CORE_delivery_update(void *context, qdr_delivery_t *dlv, uint64_t disp, bool settled) | |
| { | |
| qd_router_t *router = (qd_router_t*) context; | |
| pn_delivery_t *pnd = qdr_node_delivery_pn_from_qdr(dlv); | |
| if (!pnd) | |
| return; | |
| // If the delivery's link is somehow gone (maybe because of a connection drop, we don't proceed. | |
| if (!pn_delivery_link(pnd)) | |
| return; | |
| qdr_link_t *qlink = qdr_delivery_link(dlv); | |
| qd_link_t *link = 0; | |
| qd_connection_t *qd_conn = 0; | |
| if (qlink) { | |
| link = (qd_link_t*) qdr_link_get_context(qlink); | |
| if (link) { | |
| qd_conn = qd_link_connection(link); | |
| if (qd_conn == 0) | |
| return; | |
| } | |
| else | |
| return; | |
| } | |
| else | |
| return; | |
| if (disp && !pn_delivery_settled(pnd)) { | |
| uint64_t ignore = 0; | |
| qd_delivery_state_t *dstate = qdr_delivery_take_local_delivery_state(dlv, &ignore); | |
| // update if the disposition has changed or there is new state associated with it | |
| if (disp != pn_delivery_local_state(pnd) || dstate) { | |
| // handle propagation of delivery state from qdr_delivery_t to proton: | |
| qd_delivery_write_local_state(pnd, disp, dstate); | |
| pn_delivery_update(pnd, disp); | |
| qd_delivery_state_free(dstate); | |
| if (disp == PN_MODIFIED) // @TODO(kgiusti) why do we need this??? | |
| pn_disposition_set_failed(pn_delivery_local(pnd), true); | |
| } | |
| } | |
| if (settled) { | |
| qd_message_t *msg = qdr_delivery_message(dlv); | |
| if (qd_message_receive_complete(msg)) { | |
| // | |
| // If the delivery is settled and the message has fully arrived, disconnect | |
| // the linkages and settle it in Proton now. | |
| // | |
| qdr_node_disconnect_deliveries(router->router_core, link, dlv, pnd); | |
| pn_delivery_settle(pnd); | |
| } else { | |
| if (disp == PN_RELEASED || disp == PN_MODIFIED || disp == PN_REJECTED || qdr_delivery_presettled(dlv)) { | |
| // | |
| // If the delivery is settled and it is still arriving, defer the settlement | |
| // until the content has fully arrived. For now set the disposition on the qdr_delivery | |
| // We will use this disposition later on to set the disposition on the proton delivery. | |
| // | |
| qdr_delivery_set_disposition(dlv, disp); | |
| // | |
| // We have set the message to be discarded. We will use this information | |
| // in AMQP_rx_handler to only update the disposition on the proton delivery if the message is discarded. | |
| // | |
| qd_message_set_discard(msg, true); | |
| // | |
| // If the disposition is RELEASED or MODIFIED, set the message to discard | |
| // and if it is blocked by Q2 holdoff, get the link rolling again. | |
| // | |
| qd_message_Q2_holdoff_disable(msg); | |
| } | |
| } | |
| } | |
| } | |
| QD_EXPORT void qd_router_setup_late(qd_dispatch_t *qd) | |
| { | |
| qd->router->tracemask = qd_tracemask(); | |
| qd->router->router_core = qdr_core(qd, qd->router->router_mode, qd->router->router_area, qd->router->router_id); | |
| amqp_direct_adaptor = qdr_protocol_adaptor(qd->router->router_core, | |
| "amqp", | |
| (void*) qd->router, | |
| CORE_connection_activate, | |
| CORE_link_first_attach, | |
| CORE_link_second_attach, | |
| CORE_link_detach, | |
| CORE_link_flow, | |
| CORE_link_offer, | |
| CORE_link_drained, | |
| CORE_link_drain, | |
| CORE_link_push, | |
| CORE_link_deliver, | |
| CORE_link_get_credit, | |
| CORE_delivery_update, | |
| CORE_close_connection, | |
| CORE_conn_trace); | |
| qd_router_python_setup(qd->router); | |
| qd_timer_schedule(qd->router->timer, 1000); | |
| } | |
| void qd_router_free(qd_router_t *router) | |
| { | |
| if (!router) return; | |
| // | |
| // The char* router->router_id and router->router_area are owned by qd->router_id and qd->router_area respectively | |
| // We will set them to zero here just in case anybody tries to use these fields. | |
| // | |
| router->router_id = 0; | |
| router->router_area = 0; | |
| qd_container_set_default_node_type(router->qd, 0, 0, QD_DIST_BOTH); | |
| qdr_core_free(router->router_core); | |
| qd_tracemask_free(router->tracemask); | |
| qd_timer_free(router->timer); | |
| sys_mutex_free(router->lock); | |
| qd_router_configure_free(router); | |
| qd_router_python_free(router); | |
| free(router); | |
| qd_router_id_finalize(); | |
| free(direct_prefix); | |
| } | |
| const char *qd_router_id(void) | |
| { | |
| assert(node_id); | |
| return node_id; | |
| } | |
| const uint8_t *qd_router_id_encoded(size_t *len) | |
| { | |
| assert(encoded_node_id && encoded_node_id_len); | |
| *len = encoded_node_id_len; | |
| return encoded_node_id; | |
| } | |
| qdr_core_t *qd_router_core(qd_dispatch_t *qd) | |
| { | |
| return qd->router->router_core; | |
| } | |
| // invoked by an I/O thread when enough buffers have been released deactivate | |
| // the Q2 block. Note that this method will likely be running on a worker | |
| // thread that is not the same thread that "owns" the qd_link_t passed in. | |
| // | |
| void qd_link_q2_restart_receive(qd_alloc_safe_ptr_t context) | |
| { | |
| qd_link_t *in_link = (qd_link_t*) qd_alloc_deref_safe_ptr(&context); | |
| if (!in_link) | |
| return; | |
| assert(qd_link_direction(in_link) == QD_INCOMING); | |
| qd_connection_t *in_conn = qd_link_connection(in_link); | |
| if (in_conn) { | |
| qd_link_t_sp *safe_ptr = NEW(qd_alloc_safe_ptr_t); | |
| *safe_ptr = context; // use original to keep old sequence counter | |
| qd_connection_invoke_deferred(in_conn, deferred_AMQP_rx_handler, safe_ptr); | |
| } | |
| } | |
| // Issue a warning POLICY log message with connection and link identities | |
| // prepended to the policy denial text string. | |
| void qd_connection_log_policy_denial(qd_link_t *link, const char *text) | |
| { | |
| qdr_link_t *rlink = (qdr_link_t*) qd_link_get_context(link); | |
| uint64_t l_id = 0; | |
| uint64_t c_id = 0; | |
| if (rlink) { | |
| l_id = rlink->identity; | |
| if (rlink->conn) { | |
| c_id = rlink->conn->identity; | |
| } | |
| } | |
| qd_log(qd_policy_log_source(), QD_LOG_WARNING, "[C%"PRIu64"][L%"PRIu64"] %s", | |
| c_id, l_id, text); | |
| } |