Function clusterer_bcast_msg

modules/clusterer/clusterer.c:459–553

}

static enum clusterer_send_ret
clusterer_bcast_msg(bin_packet_t *packet, int dst_cid,
                    enum cl_node_match_op match_op, int check_cap)
{
	node_info_t *node;
	int rc, sent = 0, down = 1, matched_once = 0;
	cluster_info_t *dst_cl;
	int ev_actions_required = 0;
	str capability;

	if (!cl_list_lock) {
		LM_ERR("cluster shutdown - cannot send new messages!\n");
		return CLUSTERER_CURR_DISABLED;
	}

	lock_start_read(cl_list_lock);

	dst_cl = get_cluster_by_id(dst_cid);
	if (!dst_cl) {
		LM_ERR("Unknown cluster, id [%d]\n", dst_cid);
		lock_stop_read(cl_list_lock);
		return CLUSTERER_SEND_ERR;
	}

	lock_get(dst_cl->current_node->lock);
	if (!(dst_cl->current_node->flags & NODE_STATE_ENABLED)) {
		lock_release(dst_cl->current_node->lock);
		lock_stop_read(cl_list_lock);
		return CLUSTERER_CURR_DISABLED;
	}
	lock_release(dst_cl->current_node->lock);

	if (check_cap) {
		bin_get_capability(packet, &capability);
		rc = get_capability_status(dst_cl, &capability);
		if (rc == -1) {
			lock_stop_read(cl_list_lock);
			return CLUSTERER_SEND_ERR;
		} else if (rc == CAP_DISABLED) {
			lock_stop_read(cl_list_lock);
			LM_DBG("capability [%.*s] disabled, skip message sending\n",
				capability.len, capability.s);
			return CLUSTERER_SEND_SUCCESS;
		}
	}

	for (node = dst_cl->node_list; node; node = node->next) {
		if (!match_node(dst_cl->current_node, node, match_op))
			continue;

		lock_get(node->lock);
		if (!(node->flags & NODE_STATE_ENABLED)) {
			lock_release(node->lock);
			LM_DBG("node disabled, skip message sending\n");
			continue;
		}
		lock_release(node->lock);