| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| #include <stdio.h> |
| #include <string.h> |
| #include <qpid/dispatch/container.h> |
| #include <qpid/dispatch/message.h> |
| #include <proton/engine.h> |
| #include <proton/message.h> |
| #include <qpid/dispatch/ctools.h> |
| #include <qpid/dispatch/hash.h> |
| #include <qpid/dispatch/threading.h> |
| #include <qpid/dispatch/iterator.h> |
| #include <qpid/dispatch/log.h> |
| |
| static char *module="CONTAINER"; |
| |
| struct dx_node_t { |
| const dx_node_type_t *ntype; |
| char *name; |
| void *context; |
| dx_dist_mode_t supported_dist; |
| dx_lifetime_policy_t life_policy; |
| }; |
| |
| ALLOC_DECLARE(dx_node_t); |
| ALLOC_DEFINE(dx_node_t); |
| ALLOC_DEFINE(dx_link_item_t); |
| |
| struct dx_link_t { |
| pn_link_t *pn_link; |
| void *context; |
| dx_node_t *node; |
| }; |
| |
| ALLOC_DECLARE(dx_link_t); |
| ALLOC_DEFINE(dx_link_t); |
| |
| typedef struct nxc_node_type_t { |
| DEQ_LINKS(struct nxc_node_type_t); |
| const dx_node_type_t *ntype; |
| } nxc_node_type_t; |
| DEQ_DECLARE(nxc_node_type_t, nxc_node_type_list_t); |
| |
| |
| static hash_t *node_type_map; |
| static hash_t *node_map; |
| static sys_mutex_t *lock; |
| static dx_node_t *default_node; |
| static nxc_node_type_list_t node_type_list; |
| |
| static void setup_outgoing_link(pn_link_t *pn_link) |
| { |
| sys_mutex_lock(lock); |
| dx_node_t *node; |
| int result; |
| const char *source = pn_terminus_get_address(pn_link_remote_source(pn_link)); |
| dx_field_iterator_t *iter; |
| // TODO - Extract the name from the structured source |
| |
| if (source) { |
| iter = dx_field_iterator_string(source, ITER_VIEW_NODE_ID); |
| result = hash_retrieve(node_map, iter, (void*) &node); |
| dx_field_iterator_free(iter); |
| } else |
| result = -1; |
| sys_mutex_unlock(lock); |
| |
| if (result < 0) { |
| if (default_node) |
| node = default_node; |
| else { |
| // Reject the link |
| // TODO - When the API allows, add an error message for "no available node" |
| pn_link_close(pn_link); |
| return; |
| } |
| } |
| |
| dx_link_t *link = new_dx_link_t(); |
| if (!link) { |
| pn_link_close(pn_link); |
| return; |
| } |
| |
| link->pn_link = pn_link; |
| link->context = 0; |
| link->node = node; |
| |
| pn_link_set_context(pn_link, link); |
| node->ntype->outgoing_handler(node->context, link); |
| } |
| |
| |
| static void setup_incoming_link(pn_link_t *pn_link) |
| { |
| sys_mutex_lock(lock); |
| dx_node_t *node; |
| int result; |
| const char *target = pn_terminus_get_address(pn_link_remote_target(pn_link)); |
| dx_field_iterator_t *iter; |
| // TODO - Extract the name from the structured target |
| |
| if (target) { |
| iter = dx_field_iterator_string(target, ITER_VIEW_NODE_ID); |
| result = hash_retrieve(node_map, iter, (void*) &node); |
| dx_field_iterator_free(iter); |
| } else |
| result = -1; |
| sys_mutex_unlock(lock); |
| |
| if (result < 0) { |
| if (default_node) |
| node = default_node; |
| else { |
| // Reject the link |
| // TODO - When the API allows, add an error message for "no available node" |
| pn_link_close(pn_link); |
| return; |
| } |
| } |
| |
| dx_link_t *link = new_dx_link_t(); |
| if (!link) { |
| pn_link_close(pn_link); |
| return; |
| } |
| |
| link->pn_link = pn_link; |
| link->context = 0; |
| link->node = node; |
| |
| pn_link_set_context(pn_link, link); |
| node->ntype->incoming_handler(node->context, link); |
| } |
| |
| |
| static int do_writable(pn_link_t *pn_link) |
| { |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| if (!link) |
| return 0; |
| |
| dx_node_t *node = link->node; |
| if (!node) |
| return 0; |
| |
| return node->ntype->writable_handler(node->context, link); |
| } |
| |
| |
| static void process_receive(pn_delivery_t *delivery) |
| { |
| pn_link_t *pn_link = pn_delivery_link(delivery); |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| |
| if (link) { |
| dx_node_t *node = link->node; |
| if (node) { |
| node->ntype->rx_handler(node->context, link, delivery); |
| return; |
| } |
| } |
| |
| // |
| // Reject the delivery if we couldn't find a node to handle it |
| // |
| pn_link_advance(pn_link); |
| pn_link_flow(pn_link, 1); |
| pn_delivery_update(delivery, PN_REJECTED); |
| pn_delivery_settle(delivery); |
| } |
| |
| |
| static void do_send(pn_delivery_t *delivery) |
| { |
| pn_link_t *pn_link = pn_delivery_link(delivery); |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| |
| if (link) { |
| dx_node_t *node = link->node; |
| if (node) { |
| node->ntype->tx_handler(node->context, link, delivery); |
| return; |
| } |
| } |
| |
| // TODO - Cancel the delivery |
| } |
| |
| |
| static void do_updated(pn_delivery_t *delivery) |
| { |
| pn_link_t *pn_link = pn_delivery_link(delivery); |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| |
| if (link) { |
| dx_node_t *node = link->node; |
| if (node) |
| node->ntype->disp_handler(node->context, link, delivery); |
| } |
| } |
| |
| |
| static int close_handler(void* unused, pn_connection_t *conn) |
| { |
| // |
| // Close all links, passing False as the 'closed' argument. These links are not |
| // being properly 'detached'. They are being orphaned. |
| // |
| pn_link_t *pn_link = pn_link_head(conn, 0); |
| while (pn_link) { |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| dx_node_t *node = link->node; |
| if (node) |
| node->ntype->link_detach_handler(node->context, link, 0); |
| pn_link_close(pn_link); |
| free_dx_link_t(link); |
| pn_link = pn_link_next(pn_link, 0); |
| } |
| |
| // teardown all sessions |
| pn_session_t *ssn = pn_session_head(conn, 0); |
| while (ssn) { |
| pn_session_close(ssn); |
| ssn = pn_session_next(ssn, 0); |
| } |
| |
| // teardown the connection |
| pn_connection_close(conn); |
| return 0; |
| } |
| |
| |
| static int process_handler(void* unused, pn_connection_t *conn) |
| { |
| pn_session_t *ssn; |
| pn_link_t *pn_link; |
| pn_delivery_t *delivery; |
| int event_count = 0; |
| |
| // Step 1: setup the engine's connection, and any sessions and links |
| // that may be pending. |
| |
| // initialize the connection if it's new |
| if (pn_connection_state(conn) & PN_LOCAL_UNINIT) { |
| pn_connection_open(conn); |
| event_count++; |
| } |
| |
| // open all pending sessions |
| ssn = pn_session_head(conn, PN_LOCAL_UNINIT); |
| while (ssn) { |
| pn_session_open(ssn); |
| ssn = pn_session_next(ssn, PN_LOCAL_UNINIT); |
| event_count++; |
| } |
| |
| // configure and open any pending links |
| pn_link = pn_link_head(conn, PN_LOCAL_UNINIT); |
| while (pn_link) { |
| if (pn_link_is_sender(pn_link)) |
| setup_outgoing_link(pn_link); |
| else |
| setup_incoming_link(pn_link); |
| pn_link = pn_link_next(pn_link, PN_LOCAL_UNINIT); |
| event_count++; |
| } |
| |
| |
| // Step 2: Now drain all the pending deliveries from the connection's |
| // work queue and process them |
| |
| delivery = pn_work_head(conn); |
| while (delivery) { |
| if (pn_delivery_readable(delivery)) |
| process_receive(delivery); |
| else if (pn_delivery_writable(delivery)) |
| do_send(delivery); |
| |
| if (pn_delivery_updated(delivery)) |
| do_updated(delivery); |
| |
| delivery = pn_work_next(delivery); |
| event_count++; |
| } |
| |
| // |
| // Step 2.5: Traverse all of the links on the connection looking for |
| // outgoing links with non-zero credit. Call the attached node's |
| // writable handler for such links. |
| // |
| pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE); |
| while (pn_link) { |
| assert(pn_session_connection(pn_link_session(pn_link)) == conn); |
| if (pn_link_is_sender(pn_link) && pn_link_credit(pn_link) > 0) |
| event_count += do_writable(pn_link); |
| pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE); |
| } |
| |
| // Step 3: Clean up any links or sessions that have been closed by the |
| // remote. If the connection has been closed remotely, clean that up |
| // also. |
| |
| // teardown any terminating links |
| pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); |
| while (pn_link) { |
| dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); |
| dx_node_t *node = link->node; |
| if (node) |
| node->ntype->link_detach_handler(node->context, link, 1); // TODO - get 'closed' from detach message |
| pn_link_close(pn_link); |
| pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); |
| event_count++; |
| } |
| |
| // teardown any terminating sessions |
| ssn = pn_session_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); |
| while (ssn) { |
| pn_session_close(ssn); |
| ssn = pn_session_next(ssn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); |
| event_count++; |
| } |
| |
| // teardown the connection if it's terminating |
| if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) { |
| pn_connection_close(conn); |
| event_count++; |
| } |
| |
| return event_count; |
| } |
| |
| |
| static void open_handler(dx_connection_t *conn, dx_direction_t dir) |
| { |
| const dx_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(lock); |
| nxc_node_type_t *nt_item = DEQ_HEAD(node_type_list); |
| sys_mutex_unlock(lock); |
| |
| pn_connection_open(dx_connection_pn(conn)); |
| |
| while (nt_item) { |
| nt = nt_item->ntype; |
| if (dir == DX_INCOMING) { |
| if (nt->inbound_conn_open_handler) |
| nt->inbound_conn_open_handler(nt->type_context, conn); |
| } else { |
| if (nt->outbound_conn_open_handler) |
| nt->outbound_conn_open_handler(nt->type_context, conn); |
| } |
| |
| sys_mutex_lock(lock); |
| nt_item = DEQ_NEXT(nt_item); |
| sys_mutex_unlock(lock); |
| } |
| } |
| |
| |
| static int handler(void* context, dx_conn_event_t event, dx_connection_t *dx_conn) |
| { |
| pn_connection_t *conn = dx_connection_pn(dx_conn); |
| |
| switch (event) { |
| case DX_CONN_EVENT_LISTENER_OPEN: open_handler(dx_conn, DX_INCOMING); break; |
| case DX_CONN_EVENT_CONNECTOR_OPEN: open_handler(dx_conn, DX_OUTGOING); break; |
| case DX_CONN_EVENT_CLOSE: return close_handler(context, conn); |
| case DX_CONN_EVENT_PROCESS: return process_handler(context, conn); |
| } |
| |
| return 0; |
| } |
| |
| |
| void dx_container_initialize(void) |
| { |
| dx_log(module, LOG_TRACE, "Container Initializing"); |
| |
| node_type_map = hash(6, 4, 1); // 64 buckets, item batches of 4 |
| node_map = hash(10, 32, 0); // 1K buckets, item batches of 32 |
| lock = sys_mutex(); |
| default_node = 0; |
| DEQ_INIT(node_type_list); |
| |
| dx_server_set_conn_handler(handler); |
| } |
| |
| |
| void dx_container_finalize(void) |
| { |
| } |
| |
| |
| int dx_container_register_node_type(const dx_node_type_t *nt) |
| { |
| int result; |
| dx_field_iterator_t *iter = dx_field_iterator_string(nt->type_name, ITER_VIEW_ALL); |
| nxc_node_type_t *nt_item = NEW(nxc_node_type_t); |
| DEQ_ITEM_INIT(nt_item); |
| nt_item->ntype = nt; |
| |
| sys_mutex_lock(lock); |
| result = hash_insert_const(node_type_map, iter, nt); |
| DEQ_INSERT_TAIL(node_type_list, nt_item); |
| sys_mutex_unlock(lock); |
| |
| dx_field_iterator_free(iter); |
| if (result < 0) |
| return result; |
| dx_log(module, LOG_TRACE, "Node Type Registered - %s", nt->type_name); |
| |
| return 0; |
| } |
| |
| |
| void dx_container_set_default_node_type(const dx_node_type_t *nt, |
| void *context, |
| dx_dist_mode_t supported_dist) |
| { |
| if (default_node) |
| dx_container_destroy_node(default_node); |
| |
| if (nt) { |
| default_node = dx_container_create_node(nt, 0, context, supported_dist, DX_LIFE_PERMANENT); |
| dx_log(module, LOG_TRACE, "Node of type '%s' installed as default node", nt->type_name); |
| } else { |
| default_node = 0; |
| dx_log(module, LOG_TRACE, "Default node removed"); |
| } |
| } |
| |
| |
| dx_node_t *dx_container_create_node(const dx_node_type_t *nt, |
| const char *name, |
| void *context, |
| dx_dist_mode_t supported_dist, |
| dx_lifetime_policy_t life_policy) |
| { |
| int result; |
| dx_node_t *node = new_dx_node_t(); |
| if (!node) |
| return 0; |
| |
| node->ntype = nt; |
| node->name = 0; |
| node->context = context; |
| node->supported_dist = supported_dist; |
| node->life_policy = life_policy; |
| |
| if (name) { |
| dx_field_iterator_t *iter = dx_field_iterator_string(name, ITER_VIEW_ALL); |
| sys_mutex_lock(lock); |
| result = hash_insert(node_map, iter, node); |
| sys_mutex_unlock(lock); |
| dx_field_iterator_free(iter); |
| if (result < 0) { |
| free_dx_node_t(node); |
| return 0; |
| } |
| |
| node->name = (char*) malloc(strlen(name) + 1); |
| strcpy(node->name, name); |
| } |
| |
| if (name) |
| dx_log(module, LOG_TRACE, "Node of type '%s' created with name '%s'", nt->type_name, name); |
| |
| return node; |
| } |
| |
| |
| void dx_container_destroy_node(dx_node_t *node) |
| { |
| if (node->name) { |
| dx_field_iterator_t *iter = dx_field_iterator_string(node->name, ITER_VIEW_ALL); |
| sys_mutex_lock(lock); |
| hash_remove(node_map, iter); |
| sys_mutex_unlock(lock); |
| dx_field_iterator_free(iter); |
| free(node->name); |
| } |
| |
| free_dx_node_t(node); |
| } |
| |
| |
| void dx_container_node_set_context(dx_node_t *node, void *node_context) |
| { |
| node->context = node_context; |
| } |
| |
| |
| dx_dist_mode_t dx_container_node_get_dist_modes(const dx_node_t *node) |
| { |
| return node->supported_dist; |
| } |
| |
| |
| dx_lifetime_policy_t dx_container_node_get_life_policy(const dx_node_t *node) |
| { |
| return node->life_policy; |
| } |
| |
| |
| dx_link_t *dx_link(dx_node_t *node, dx_connection_t *conn, dx_direction_t dir, const char* name) |
| { |
| pn_session_t *sess = pn_session(dx_connection_pn(conn)); |
| dx_link_t *link = new_dx_link_t(); |
| |
| if (dir == DX_OUTGOING) |
| link->pn_link = pn_sender(sess, name); |
| else |
| link->pn_link = pn_receiver(sess, name); |
| link->context = node->context; |
| link->node = node; |
| |
| pn_link_set_context(link->pn_link, link); |
| |
| pn_session_open(sess); |
| |
| return link; |
| } |
| |
| |
| void dx_link_set_context(dx_link_t *link, void *context) |
| { |
| link->context = context; |
| } |
| |
| |
| void *dx_link_get_context(dx_link_t *link) |
| { |
| return link->context; |
| } |
| |
| |
| pn_link_t *dx_link_pn(dx_link_t *link) |
| { |
| return link->pn_link; |
| } |
| |
| |
| pn_terminus_t *dx_link_source(dx_link_t *link) |
| { |
| return pn_link_source(link->pn_link); |
| } |
| |
| |
| pn_terminus_t *dx_link_target(dx_link_t *link) |
| { |
| return pn_link_target(link->pn_link); |
| } |
| |
| |
| pn_terminus_t *dx_link_remote_source(dx_link_t *link) |
| { |
| return pn_link_remote_source(link->pn_link); |
| } |
| |
| |
| pn_terminus_t *dx_link_remote_target(dx_link_t *link) |
| { |
| return pn_link_remote_target(link->pn_link); |
| } |
| |
| |
| void dx_link_activate(dx_link_t *link) |
| { |
| if (!link || !link->pn_link) |
| return; |
| |
| pn_session_t *sess = pn_link_session(link->pn_link); |
| if (!sess) |
| return; |
| |
| pn_connection_t *conn = pn_session_connection(sess); |
| if (!conn) |
| return; |
| |
| dx_connection_t *ctx = pn_connection_get_context(conn); |
| if (!ctx) |
| return; |
| |
| dx_server_activate(ctx); |
| } |
| |
| |
| void dx_link_close(dx_link_t *link) |
| { |
| pn_link_close(link->pn_link); |
| } |
| |
| |
