| /* |
| * 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 "agent_link.h" |
| #include <inttypes.h> |
| #include <stdio.h> |
| |
| #define QDR_LINK_NAME 0 |
| #define QDR_LINK_IDENTITY 1 |
| #define QDR_LINK_TYPE 2 |
| #define QDR_LINK_LINK_NAME 3 |
| #define QDR_LINK_LINK_TYPE 4 |
| #define QDR_LINK_LINK_DIR 5 |
| #define QDR_LINK_OWNING_ADDR 6 |
| #define QDR_LINK_CAPACITY 7 |
| #define QDR_LINK_PEER 8 |
| #define QDR_LINK_UNDELIVERED_COUNT 9 |
| #define QDR_LINK_UNSETTLED_COUNT 10 |
| #define QDR_LINK_DELIVERY_COUNT 11 |
| #define QDR_LINK_CONNECTION_ID 12 |
| #define QDR_LINK_ADMIN_STATE 13 |
| #define QDR_LINK_OPER_STATE 14 |
| #define QDR_LINK_PRESETTLED_COUNT 15 |
| #define QDR_LINK_DROPPED_PRESETTLED_COUNT 16 |
| #define QDR_LINK_ACCEPTED_COUNT 17 |
| #define QDR_LINK_REJECTED_COUNT 18 |
| #define QDR_LINK_RELEASED_COUNT 19 |
| #define QDR_LINK_MODIFIED_COUNT 20 |
| #define QDR_LINK_DELAYED_1SEC 21 |
| #define QDR_LINK_DELAYED_10SEC 22 |
| #define QDR_LINK_DELIVERIES_STUCK 23 |
| #define QDR_LINK_INGRESS_HISTOGRAM 24 |
| #define QDR_LINK_PRIORITY 25 |
| #define QDR_LINK_SETTLE_RATE 26 |
| #define QDR_LINK_CREDIT_AVAILABLE 27 |
| #define QDR_LINK_ZERO_CREDIT_SECONDS 28 |
| |
| const char *qdr_link_columns[] = |
| {"name", |
| "identity", |
| "type", |
| "linkName", |
| "linkType", |
| "linkDir", |
| "owningAddr", |
| "capacity", |
| "peer", |
| "undeliveredCount", |
| "unsettledCount", |
| "deliveryCount", |
| "connectionId", // The connection id of the owner connection |
| "adminStatus", |
| "operStatus", |
| "presettledCount", |
| "droppedPresettledCount", |
| "acceptedCount", |
| "rejectedCount", |
| "releasedCount", |
| "modifiedCount", |
| "deliveriesDelayed1Sec", |
| "deliveriesDelayed10Sec", |
| "deliveriesStuck", |
| "ingressHistogram", |
| "priority", |
| "settleRate", |
| "creditAvailable", |
| "zeroCreditSeconds", |
| 0}; |
| |
| static const char *qd_link_type_name(qd_link_type_t lt) |
| { |
| switch (lt) { |
| case QD_LINK_ENDPOINT : return "endpoint"; |
| case QD_LINK_CONTROL : return "router-control"; |
| case QD_LINK_ROUTER : return "inter-router"; |
| case QD_LINK_EDGE_DOWNLINK : return "edge-downlink"; |
| } |
| |
| return ""; |
| } |
| |
| |
| static const char *address_key(qdr_address_t *addr) |
| { |
| return addr && addr->hash_handle ? (const char*) qd_hash_key_by_handle(addr->hash_handle) : NULL; |
| } |
| |
| static void qdr_agent_write_column_CT(qdr_core_t *core, qd_composed_field_t *body, int col, qdr_link_t *link) |
| { |
| char *text = 0; |
| |
| if (!link) |
| return; |
| |
| switch(col) { |
| case QDR_LINK_NAME: { |
| if (link->name) |
| qd_compose_insert_string(body, link->name); |
| else |
| qd_compose_insert_null(body); |
| break; |
| } |
| |
| case QDR_LINK_IDENTITY: { |
| char id[100]; |
| snprintf(id, 100, "%"PRId64, link->identity); |
| qd_compose_insert_string(body, id); |
| break; |
| } |
| |
| case QDR_LINK_TYPE: |
| qd_compose_insert_string(body, "org.apache.qpid.dispatch.router.link"); |
| break; |
| |
| case QDR_LINK_LINK_NAME: |
| qd_compose_insert_string(body, link->name); |
| break; |
| |
| case QDR_LINK_LINK_TYPE: |
| qd_compose_insert_string(body, qd_link_type_name(link->link_type)); |
| break; |
| |
| case QDR_LINK_LINK_DIR: |
| qd_compose_insert_string(body, link->link_direction == QD_INCOMING ? "in" : "out"); |
| break; |
| |
| case QDR_LINK_OWNING_ADDR: |
| if(link->owning_addr) |
| qd_compose_insert_string(body, address_key(link->owning_addr)); |
| else if (link->connected_link && link->connected_link->terminus_addr) |
| qd_compose_insert_string(body, link->connected_link->terminus_addr); |
| else if (link->terminus_addr) |
| qd_compose_insert_string(body, link->terminus_addr); |
| else |
| qd_compose_insert_null(body); |
| break; |
| |
| case QDR_LINK_CAPACITY: |
| qd_compose_insert_uint(body, link->capacity); |
| break; |
| |
| case QDR_LINK_PEER: |
| if (link->connected_link) { |
| char id[100]; |
| snprintf(id, 100, "%"PRId64, link->connected_link->identity); |
| qd_compose_insert_string(body, id); |
| } else |
| qd_compose_insert_null(body); |
| break; |
| |
| case QDR_LINK_UNDELIVERED_COUNT: |
| qd_compose_insert_ulong(body, DEQ_SIZE(link->undelivered)); |
| break; |
| |
| case QDR_LINK_UNSETTLED_COUNT: |
| qd_compose_insert_ulong(body, DEQ_SIZE(link->unsettled)); |
| break; |
| |
| case QDR_LINK_DELIVERY_COUNT: |
| qd_compose_insert_ulong(body, link->total_deliveries); |
| break; |
| |
| case QDR_LINK_CONNECTION_ID: { |
| char id[100]; |
| snprintf(id, 100, "%"PRId64, link->conn->identity); |
| qd_compose_insert_string(body, id); |
| break; |
| } |
| |
| case QDR_LINK_ADMIN_STATE: |
| text = link->admin_enabled ? "enabled" : "disabled"; |
| qd_compose_insert_string(body, text); |
| break; |
| |
| case QDR_LINK_OPER_STATE: |
| switch (link->oper_status) { |
| case QDR_LINK_OPER_UP: text = "up"; break; |
| case QDR_LINK_OPER_DOWN: text = "down"; break; |
| case QDR_LINK_OPER_QUIESCING: text = "quiescing"; break; |
| case QDR_LINK_OPER_IDLE: text = "idle"; break; |
| default: |
| text = 0; |
| } |
| if (!!text) |
| qd_compose_insert_string(body, text); |
| else |
| qd_compose_insert_null(body); |
| break; |
| |
| case QDR_LINK_PRESETTLED_COUNT: |
| qd_compose_insert_ulong(body, link->presettled_deliveries); |
| break; |
| |
| case QDR_LINK_DROPPED_PRESETTLED_COUNT: { |
| qd_compose_insert_ulong(body, link->dropped_presettled_deliveries); |
| break; |
| } |
| |
| case QDR_LINK_ACCEPTED_COUNT: |
| qd_compose_insert_ulong(body, link->accepted_deliveries); |
| break; |
| |
| case QDR_LINK_REJECTED_COUNT: |
| qd_compose_insert_ulong(body, link->rejected_deliveries); |
| break; |
| |
| case QDR_LINK_RELEASED_COUNT: |
| qd_compose_insert_ulong(body, link->released_deliveries); |
| break; |
| |
| case QDR_LINK_MODIFIED_COUNT: |
| qd_compose_insert_ulong(body, link->modified_deliveries); |
| break; |
| |
| case QDR_LINK_DELAYED_1SEC: |
| qd_compose_insert_ulong(body, link->deliveries_delayed_1sec); |
| break; |
| |
| case QDR_LINK_DELAYED_10SEC: |
| qd_compose_insert_ulong(body, link->deliveries_delayed_10sec); |
| break; |
| |
| case QDR_LINK_DELIVERIES_STUCK: |
| qd_compose_insert_ulong(body, link->deliveries_stuck); |
| break; |
| |
| case QDR_LINK_INGRESS_HISTOGRAM: |
| if (link->ingress_histogram) { |
| qd_compose_start_list(body); |
| for (int i = 0; i < qd_bitmask_width(); i++) |
| qd_compose_insert_ulong(body, link->ingress_histogram[i]); |
| qd_compose_end_list(body); |
| } else |
| qd_compose_insert_null(body); |
| break; |
| |
| case QDR_LINK_PRIORITY: |
| qd_compose_insert_uint(body, link->priority); |
| break; |
| |
| case QDR_LINK_SETTLE_RATE: { |
| uint32_t delta_time = core->uptime_ticks - link->core_ticks; |
| if (delta_time > 0) { |
| if (delta_time > QDR_LINK_RATE_DEPTH) |
| delta_time = QDR_LINK_RATE_DEPTH; |
| for (uint8_t delta_slots = 0; delta_slots < delta_time; delta_slots++) { |
| link->rate_cursor = (link->rate_cursor + 1) % QDR_LINK_RATE_DEPTH; |
| link->settled_deliveries[link->rate_cursor] = 0; |
| } |
| link->core_ticks = core->uptime_ticks; |
| } |
| |
| uint64_t total = 0; |
| for (uint8_t i = 0; i < QDR_LINK_RATE_DEPTH; i++) |
| total += link->settled_deliveries[i]; |
| qd_compose_insert_uint(body, total / QDR_LINK_RATE_DEPTH); |
| } |
| break; |
| |
| case QDR_LINK_CREDIT_AVAILABLE: |
| qd_compose_insert_uint(body, link->credit_reported); |
| break; |
| |
| case QDR_LINK_ZERO_CREDIT_SECONDS: |
| if (link->zero_credit_time == 0) |
| qd_compose_insert_uint(body, 0); |
| else |
| qd_compose_insert_uint(body, core->uptime_ticks - link->zero_credit_time); |
| break; |
| |
| default: |
| qd_compose_insert_null(body); |
| break; |
| } |
| } |
| |
| static void qdr_agent_write_link_CT(qdr_core_t *core, qdr_query_t *query, qdr_link_t *link) |
| { |
| qd_composed_field_t *body = query->body; |
| |
| qd_compose_start_list(body); |
| if (link) { |
| int i = 0; |
| while (query->columns[i] >= 0) { |
| qdr_agent_write_column_CT(core, body, query->columns[i], link); |
| i++; |
| } |
| } |
| qd_compose_end_list(body); |
| } |
| |
| static void qdr_manage_advance_link_CT(qdr_query_t *query, qdr_link_t *link) |
| { |
| query->next_offset++; |
| link = DEQ_NEXT(link); |
| if (link) { |
| query->more = true; |
| //query->next_key = qdr_field((const char*) qd_hash_key_by_handle(link->owning_addr->hash_handle)); |
| } else |
| query->more = false; |
| } |
| |
| |
| void qdra_link_get_first_CT(qdr_core_t *core, qdr_query_t *query, int offset) |
| { |
| // |
| // Queries that get this far will always succeed. |
| // |
| query->status = QD_AMQP_OK; |
| |
| // |
| // If the offset goes beyond the set of links, end the query now. |
| // |
| if (offset >= DEQ_SIZE(core->open_links)) { |
| query->more = false; |
| qdr_agent_enqueue_response_CT(core, query); |
| return; |
| } |
| |
| // |
| // Run to the address at the offset. |
| // |
| qdr_link_t *link = DEQ_HEAD(core->open_links); |
| for (int i = 0; i < offset && link; i++) |
| link = DEQ_NEXT(link); |
| assert(link); |
| |
| if (link) { |
| // |
| // Write the columns of the link into the response body. |
| // |
| qdr_agent_write_link_CT(core, query, link); |
| |
| // |
| // Advance to the next address |
| // |
| query->next_offset = offset; |
| qdr_manage_advance_link_CT(query, link); |
| } |
| else { |
| query->more = false; |
| } |
| |
| // |
| // Enqueue the response. |
| // |
| qdr_agent_enqueue_response_CT(core, query); |
| } |
| |
| |
| void qdra_link_get_next_CT(qdr_core_t *core, qdr_query_t *query) |
| { |
| qdr_link_t *link = 0; |
| |
| if (query->next_offset < DEQ_SIZE(core->open_links)) { |
| link = DEQ_HEAD(core->open_links); |
| for (int i = 0; i < query->next_offset && link; i++) |
| link = DEQ_NEXT(link); |
| } |
| |
| if (link) { |
| // |
| // Write the columns of the link entity into the response body. |
| // |
| qdr_agent_write_link_CT(core, query, link); |
| |
| // |
| // Advance to the next link |
| // |
| qdr_manage_advance_link_CT(query, link); |
| } else |
| query->more = false; |
| |
| // |
| // Enqueue the response. |
| // |
| qdr_agent_enqueue_response_CT(core, query); |
| } |
| |
| |
| static void qdr_manage_write_response_map_CT(qdr_core_t *core, qd_composed_field_t *body, qdr_link_t *link) |
| { |
| qd_compose_start_map(body); |
| |
| for(int i = 0; i < QDR_LINK_COLUMN_COUNT; i++) { |
| qd_compose_insert_string(body, qdr_link_columns[i]); |
| qdr_agent_write_column_CT(core, body, i, link); |
| } |
| |
| qd_compose_end_map(body); |
| } |
| |
| |
| static qdr_link_t *qdr_link_find_by_identity(qdr_core_t *core, qd_iterator_t *identity) |
| { |
| if (!identity) |
| return 0; |
| |
| qdr_link_t *link = DEQ_HEAD(core->open_links); |
| |
| while(link) { |
| char id[100]; |
| if (link->identity) { |
| snprintf(id, 100, "%"PRId64, link->identity); |
| if (qd_iterator_equal(identity, (const unsigned char *)id)) |
| break; |
| } |
| link = DEQ_NEXT(link); |
| } |
| |
| return link; |
| |
| } |
| |
| |
| static qdr_link_t *qdr_link_find_by_name(qdr_core_t *core, qd_iterator_t *name) |
| { |
| if(!name) |
| return 0; |
| |
| qdr_link_t *link = DEQ_HEAD(core->open_links); |
| |
| while(link) { |
| if (link->name && qd_iterator_equal(name, (const unsigned char *)link->name)) |
| break; |
| link = DEQ_NEXT(link); |
| } |
| |
| return link; |
| } |
| |
| |
| static void qdra_link_update_set_status(qdr_core_t *core, qdr_query_t *query, qdr_link_t *link) |
| { |
| if (link) { |
| //link->admin_state = qd_iterator_copy(adm_state); |
| qdr_manage_write_response_map_CT(core, query->body, link); |
| query->status = QD_AMQP_OK; |
| } |
| else { |
| query->status = QD_AMQP_NOT_FOUND; |
| qd_compose_start_map(query->body); |
| qd_compose_end_map(query->body); |
| } |
| } |
| |
| static void qdra_link_set_bad_request(qdr_query_t *query) |
| { |
| query->status = QD_AMQP_BAD_REQUEST; |
| qd_compose_start_map(query->body); |
| qd_compose_end_map(query->body); |
| } |
| |
| void qdra_link_update_CT(qdr_core_t *core, |
| qd_iterator_t *name, |
| qd_iterator_t *identity, |
| qdr_query_t *query, |
| qd_parsed_field_t *in_body) |
| |
| { |
| // If the request was successful then the statusCode MUST contain 200 (OK) and the body of the message |
| // MUST contain a map containing the actual attributes of the entity updated. These MAY differ from those |
| // requested. |
| // A map containing attributes that are not applicable for the entity being created, or invalid values for a |
| // given attribute, MUST result in a failure response with a statusCode of 400 (Bad Request). |
| if (qd_parse_is_map(in_body)) { |
| // The absence of an attribute name implies that the entity should retain its existing value. |
| // If the map contains a key-value pair where the value is null then the updated entity should have no value |
| // for that attribute, removing any previous value. |
| |
| qd_parsed_field_t *admin_state = qd_parse_value_by_key(in_body, qdr_link_columns[QDR_LINK_ADMIN_STATE]); |
| if (admin_state) { //admin state is the only field that can be updated via the update management request |
| //qd_iterator_t *adm_state = qd_parse_raw(admin_state); |
| |
| if (identity) { |
| qdr_link_t *link = qdr_link_find_by_identity(core, identity); |
| // TODO - set the adm_state on the link |
| qdra_link_update_set_status(core, query, link); |
| } |
| else if (name) { |
| qdr_link_t *link = qdr_link_find_by_name(core, name); |
| // TODO - set the adm_state on the link |
| qdra_link_update_set_status(core, query, link); |
| } |
| else { |
| qdra_link_set_bad_request(query); |
| } |
| } |
| else |
| qdra_link_set_bad_request(query); |
| |
| } |
| else |
| query->status = QD_AMQP_BAD_REQUEST; |
| |
| // |
| // Enqueue the response. |
| // |
| qdr_agent_enqueue_response_CT(core, query); |
| } |
| |
