cdr.c

		return 1;
	}
	if (strcasecmp(cdr->party_a.snapshot->name, channel->name)
		&& cdr->party_b.snapshot
		&& strcasecmp(cdr->party_b.snapshot->name, channel->name)) {
		return 1;
	}
	cdr_object_transition_state(cdr, &finalized_state_fn_table);

	return 0;
}

/* PARKED STATE */

static int parked_state_process_bridge_leave(struct cdr_object *cdr, struct ast_bridge_snapshot *bridge, struct ast_channel_snapshot *channel)
{
	if (strcasecmp(cdr->party_a.snapshot->name, channel->name)) {
		return 1;
	}
	cdr_object_transition_state(cdr, &finalized_state_fn_table);

	return 0;
}

/* FINALIZED STATE */

static void finalized_state_init_function(struct cdr_object *cdr)
{
	cdr_object_finalize(cdr);
}

static int finalized_state_process_party_a(struct cdr_object *cdr, struct ast_channel_snapshot *snapshot)
{
	if (ast_test_flag(&snapshot->softhangup_flags, AST_SOFTHANGUP_HANGUP_EXEC)
		&& is_cdr_flag_set(CDR_END_BEFORE_H_EXTEN)) {
		return 0;
	}

	/* Indicate that, if possible, we should get a new CDR */
	return 1;
}

/*!
 * \internal
 * \brief Filter channel snapshots by technology
 */
static int filter_channel_snapshot(struct ast_channel_snapshot *snapshot)
{
	return snapshot->tech_properties & AST_CHAN_TP_INTERNAL;
}

/*!
 * \internal
 * \brief Filter a channel cache update
 */
static int filter_channel_cache_message(struct ast_channel_snapshot *old_snapshot,
		struct ast_channel_snapshot *new_snapshot)
{
	int ret = 0;

	/* Drop cache updates from certain channel technologies */
	if (old_snapshot) {
		ret |= filter_channel_snapshot(old_snapshot);
	}
	if (new_snapshot) {
		ret |= filter_channel_snapshot(new_snapshot);
	}

	return ret;
}

/* TOPIC ROUTER CALLBACKS */

/*!
 * \brief Handler for Stasis-Core dial messages
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message The message
 */
static void handle_dial_message(void *data, struct stasis_subscription *sub, struct stasis_message *message)
{
	struct cdr_object *cdr;
	struct ast_multi_channel_blob *payload = stasis_message_data(message);
	struct ast_channel_snapshot *caller;
	struct ast_channel_snapshot *peer;
	struct cdr_object *it_cdr;
	struct ast_json *dial_status_blob;
	const char *dial_status = NULL;
	int res = 1;

	caller = ast_multi_channel_blob_get_channel(payload, "caller");
	peer = ast_multi_channel_blob_get_channel(payload, "peer");
	if (!peer && !caller) {
		return;
	}
	if (filter_channel_snapshot(peer) || (caller && filter_channel_snapshot(caller))) {
		return;
	}

	dial_status_blob = ast_json_object_get(ast_multi_channel_blob_get_json(payload), "dialstatus");
	if (dial_status_blob) {
		dial_status = ast_json_string_get(dial_status_blob);
	}

	CDR_DEBUG("Dial %s message for %s, %s: %u.%08u\n",
		ast_strlen_zero(dial_status) ? "Begin" : "End",
		caller ? caller->name : "(none)",
		peer ? peer->name : "(none)",
		(unsigned int)stasis_message_timestamp(message)->tv_sec,
		(unsigned int)stasis_message_timestamp(message)->tv_usec);

	/* Figure out who is running this show */
	if (caller) {
		cdr = ao2_find(active_cdrs_master, caller->uniqueid, OBJ_SEARCH_KEY);
	} else {
		cdr = ao2_find(active_cdrs_master, peer->uniqueid, OBJ_SEARCH_KEY);
	}
	if (!cdr) {
		ast_log(AST_LOG_WARNING, "No CDR for channel %s\n", caller ? caller->name : peer->name);
		ast_assert(0);
		return;
	}

	ao2_lock(cdr);
	for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
		if (ast_strlen_zero(dial_status)) {
			if (!it_cdr->fn_table->process_dial_begin) {
				continue;
			}
			CDR_DEBUG("%p - Processing Dial Begin message for channel %s, peer %s\n",
				it_cdr,
				caller ? caller->name : "(none)",
				peer ? peer->name : "(none)");
			res &= it_cdr->fn_table->process_dial_begin(it_cdr,
					caller,
					peer);
		} else {
			if (!it_cdr->fn_table->process_dial_end) {
				continue;
			}
			CDR_DEBUG("%p - Processing Dial End message for channel %s, peer %s\n",
				it_cdr,
				caller ? caller->name : "(none)",
				peer ? peer->name : "(none)");
			it_cdr->fn_table->process_dial_end(it_cdr,
					caller,
					peer,
					dial_status);
		}
	}

	/* If no CDR handled a dial begin message, make a new one */
	if (res && ast_strlen_zero(dial_status)) {
		struct cdr_object *new_cdr;

		new_cdr = cdr_object_create_and_append(cdr);
		if (new_cdr) {
			new_cdr->fn_table->process_dial_begin(new_cdr, caller, peer);
		}
	}
	ao2_unlock(cdr);
	ao2_cleanup(cdr);
}

static int cdr_object_finalize_party_b(void *obj, void *arg, void *data, int flags)
{
	struct cdr_object *cdr = obj;

	if (!strcasecmp(cdr->party_b_name, arg)) {
#ifdef AST_DEVMODE
		struct ast_channel_snapshot *party_b = data;

		/*
		 * For sanity's sake we also assert the party_b snapshot
		 * is consistent with the key.
		 */
		ast_assert(cdr->party_b.snapshot
			&& !strcasecmp(cdr->party_b.snapshot->name, party_b->name));
#endif

		/* Don't transition to the finalized state - let the Party A do
		 * that when its ready
		 */
		cdr_object_finalize(cdr);
	}
	return 0;
}

static int cdr_object_update_party_b(void *obj, void *arg, void *data, int flags)
{
	struct cdr_object *cdr = obj;

	if (cdr->fn_table->process_party_b
		&& !strcasecmp(cdr->party_b_name, arg)) {
		struct ast_channel_snapshot *party_b = data;

		/*
		 * For sanity's sake we also check the party_b snapshot
		 * for consistency with the key.  The callback needs and
		 * asserts the snapshot to be this way.
		 */
		if (!cdr->party_b.snapshot
			|| strcasecmp(cdr->party_b.snapshot->name, party_b->name)) {
			ast_log(LOG_NOTICE,
				"CDR for Party A %s(%s) has inconsistent Party B %s name.  Message can be ignored but this shouldn't happen.\n",
				cdr->linkedid,
				cdr->party_a.snapshot->name,
				cdr->party_b_name);
			return 0;
		}

		cdr->fn_table->process_party_b(cdr, party_b);
	}
	return 0;
}

/*! \brief Determine if we need to add a new CDR based on snapshots */
static int check_new_cdr_needed(struct ast_channel_snapshot *old_snapshot,
		struct ast_channel_snapshot *new_snapshot)
{
	/* If we're dead, we don't need a new CDR */
	if (!new_snapshot
		|| (ast_test_flag(&new_snapshot->softhangup_flags, AST_SOFTHANGUP_HANGUP_EXEC)
			&& is_cdr_flag_set(CDR_END_BEFORE_H_EXTEN))) {
		return 0;
	}

	/* Auto-fall through will increment the priority but have no application */
	if (ast_strlen_zero(new_snapshot->appl)) {
		return 0;
	}

	if (old_snapshot && !snapshot_cep_changed(old_snapshot, new_snapshot)) {
		return 0;
	}

	return 1;
}

/*!
 * \brief Handler for Stasis-Core channel cache update messages
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message The message
 */
static void handle_channel_cache_message(void *data, struct stasis_subscription *sub, struct stasis_message *message)
{
	struct cdr_object *cdr;
	struct stasis_cache_update *update = stasis_message_data(message);
	struct ast_channel_snapshot *old_snapshot;
	struct ast_channel_snapshot *new_snapshot;
	struct cdr_object *it_cdr;

	ast_assert(update != NULL);
	ast_assert(ast_channel_snapshot_type() == update->type);

	old_snapshot = stasis_message_data(update->old_snapshot);
	new_snapshot = stasis_message_data(update->new_snapshot);

	if (filter_channel_cache_message(old_snapshot, new_snapshot)) {
		return;
	}

	if (new_snapshot && !old_snapshot) {
		cdr = cdr_object_alloc(new_snapshot);
		if (!cdr) {
			return;
		}
		cdr->is_root = 1;
		ao2_link(active_cdrs_master, cdr);
	} else {
		const char *uniqueid;

		uniqueid = new_snapshot ? new_snapshot->uniqueid : old_snapshot->uniqueid;
		cdr = ao2_find(active_cdrs_master, uniqueid, OBJ_SEARCH_KEY);
	}

	/* Handle Party A */
	if (!cdr) {
		const char *name;

		name = new_snapshot ? new_snapshot->name : old_snapshot->name;
		ast_log(AST_LOG_WARNING, "No CDR for channel %s\n", name);
		ast_assert(0);
	} else if (new_snapshot) {
		int all_reject = 1;

		ao2_lock(cdr);
		for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
			if (!it_cdr->fn_table->process_party_a) {
				continue;
			}
			all_reject &= it_cdr->fn_table->process_party_a(it_cdr, new_snapshot);
		}
		if (all_reject && check_new_cdr_needed(old_snapshot, new_snapshot)) {
			/* We're not hung up and we have a new snapshot - we need a new CDR */
			struct cdr_object *new_cdr;

			new_cdr = cdr_object_create_and_append(cdr);
			if (new_cdr) {
				new_cdr->fn_table->process_party_a(new_cdr, new_snapshot);
			}
		}
		ao2_unlock(cdr);
	} else {
		ao2_lock(cdr);
		CDR_DEBUG("%p - Beginning finalize/dispatch for %s\n", cdr, old_snapshot->name);
		for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
			cdr_object_finalize(it_cdr);
		}
		cdr_object_dispatch(cdr);
		ao2_unlock(cdr);

		cdr_all_unlink(cdr);
		ao2_unlink(active_cdrs_master, cdr);
	}

	/* Handle Party B */
	if (new_snapshot) {
		ao2_callback_data(active_cdrs_all, OBJ_NODATA | OBJ_MULTIPLE | OBJ_SEARCH_KEY,
			cdr_object_update_party_b, (char *) new_snapshot->name, new_snapshot);
	} else {
		ao2_callback_data(active_cdrs_all, OBJ_NODATA | OBJ_MULTIPLE | OBJ_SEARCH_KEY,
			cdr_object_finalize_party_b, (char *) old_snapshot->name, old_snapshot);
	}

	ao2_cleanup(cdr);
}

struct bridge_leave_data {
	struct ast_bridge_snapshot *bridge;
	struct ast_channel_snapshot *channel;
};

/*! \brief Callback used to notify CDRs of a Party B leaving the bridge */
static int cdr_object_party_b_left_bridge_cb(void *obj, void *arg, void *data, int flags)
{
	struct cdr_object *cdr = obj;
	struct bridge_leave_data *leave_data = data;

	if (cdr->fn_table == &bridge_state_fn_table
		&& !strcmp(cdr->bridge, leave_data->bridge->uniqueid)
		&& !strcasecmp(cdr->party_b_name, arg)) {
		/*
		 * For sanity's sake we also assert the party_b snapshot
		 * is consistent with the key.
		 */
		ast_assert(cdr->party_b.snapshot
			&& !strcasecmp(cdr->party_b.snapshot->name, leave_data->channel->name));

		/* It is our Party B, in our bridge. Set the end time and let the handler
		 * transition our CDR appropriately when we leave the bridge.
		 */
		cdr_object_finalize(cdr);
	}
	return 0;
}

/*! \brief Filter bridge messages based on bridge technology */
static int filter_bridge_messages(struct ast_bridge_snapshot *bridge)
{
	/* Ignore holding bridge technology messages. We treat this simply as an application
	 * that a channel enters into.
	 */
	if (!strcmp(bridge->technology, "holding_bridge") && strcmp(bridge->subclass, "parking")) {
		return 1;
	}
	return 0;
}

/*!
 * \brief Handler for when a channel leaves a bridge
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message The message - hopefully a bridge one!
 */
static void handle_bridge_leave_message(void *data, struct stasis_subscription *sub,
		struct stasis_message *message)
{
	struct ast_bridge_blob *update = stasis_message_data(message);
	struct ast_bridge_snapshot *bridge = update->bridge;
	struct ast_channel_snapshot *channel = update->channel;
	struct cdr_object *cdr;
	struct cdr_object *it_cdr;
	struct bridge_leave_data leave_data = {
		.bridge = bridge,
		.channel = channel,
	};
	int left_bridge = 0;

	if (filter_bridge_messages(bridge)) {
		return;
	}

	if (filter_channel_snapshot(channel)) {
		return;
	}

	CDR_DEBUG("Bridge Leave message for %s: %u.%08u\n",
		channel->name,
		(unsigned int)stasis_message_timestamp(message)->tv_sec,
		(unsigned int)stasis_message_timestamp(message)->tv_usec);

	cdr = ao2_find(active_cdrs_master, channel->uniqueid, OBJ_SEARCH_KEY);
	if (!cdr) {
		ast_log(AST_LOG_WARNING, "No CDR for channel %s\n", channel->name);
		ast_assert(0);
		return;
	}

	/* Party A */
	ao2_lock(cdr);
	for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
		if (!it_cdr->fn_table->process_bridge_leave) {
			continue;
		}
		CDR_DEBUG("%p - Processing Bridge Leave for %s\n",
			it_cdr, channel->name);
		if (!it_cdr->fn_table->process_bridge_leave(it_cdr, bridge, channel)) {
			ast_string_field_set(it_cdr, bridge, "");
			left_bridge = 1;
		}
	}
	ao2_unlock(cdr);

	/* Party B */
	if (left_bridge
		&& strcmp(bridge->subclass, "parking")) {
		ao2_callback_data(active_cdrs_all, OBJ_NODATA | OBJ_MULTIPLE | OBJ_SEARCH_KEY,
			cdr_object_party_b_left_bridge_cb, (char *) leave_data.channel->name,
			&leave_data);
	}

	ao2_cleanup(cdr);
}

/*!
 * \internal
 * \brief Create a new CDR, append it to an existing CDR, and update its snapshots
 *
 * \note The new CDR will be automatically transitioned to the bridge state
 */
static void bridge_candidate_add_to_cdr(struct cdr_object *cdr,
		struct cdr_object_snapshot *party_b)
{
	struct cdr_object *new_cdr;

	new_cdr = cdr_object_create_and_append(cdr);
	if (!new_cdr) {
		return;
	}
	cdr_object_snapshot_copy(&new_cdr->party_b, party_b);
	cdr_all_relink(new_cdr);
	cdr_object_check_party_a_answer(new_cdr);
	ast_string_field_set(new_cdr, bridge, cdr->bridge);
	cdr_object_transition_state(new_cdr, &bridge_state_fn_table);
	CDR_DEBUG("%p - Party A %s has new Party B %s\n",
		new_cdr, new_cdr->party_a.snapshot->name,
		party_b->snapshot->name);
}

/*!
 * \brief Process a single \ref bridge_candidate
 *
 * When a CDR enters a bridge, it needs to make pairings with everyone else
 * that it is not currently paired with. This function determines, for the
 * CDR for the channel that entered the bridge and the CDR for every other
 * channel currently in the bridge, who is Party A and makes new CDRs.
 *
 * \param cdr The \ref cdr_obj being processed
 * \param cand_cdr The \ref cdr_object that is a candidate
 *
 */
static void bridge_candidate_process(struct cdr_object *cdr, struct cdr_object *base_cand_cdr)
{
	struct cdr_object_snapshot *party_a;
	struct cdr_object *cand_cdr;

	ao2_lock(base_cand_cdr);

	for (cand_cdr = base_cand_cdr; cand_cdr; cand_cdr = cand_cdr->next) {
		/* Skip any records that are not in this bridge */
		if (strcmp(cand_cdr->bridge, cdr->bridge)) {
			continue;
		}

		/* If the candidate is us or someone we've taken on, pass on by */
		if (!strcasecmp(cdr->party_a.snapshot->name, cand_cdr->party_a.snapshot->name)
			|| (cdr->party_b.snapshot
				&& !strcasecmp(cdr->party_b.snapshot->name, cand_cdr->party_a.snapshot->name))) {
			break;
		}

		party_a = cdr_object_pick_party_a(&cdr->party_a, &cand_cdr->party_a);
		/* We're party A - make a new CDR, append it to us, and set the candidate as
		 * Party B */
		if (!strcasecmp(party_a->snapshot->name, cdr->party_a.snapshot->name)) {
			bridge_candidate_add_to_cdr(cdr, &cand_cdr->party_a);
			break;
		}

		/* We're Party B. Check if we can add ourselves immediately or if we need
		 * a new CDR for them (they already have a Party B) */
		if (cand_cdr->party_b.snapshot
			&& strcasecmp(cand_cdr->party_b.snapshot->name, cdr->party_a.snapshot->name)) {
			bridge_candidate_add_to_cdr(cand_cdr, &cdr->party_a);
		} else {
			CDR_DEBUG("%p - Party A %s has new Party B %s\n",
				cand_cdr, cand_cdr->party_a.snapshot->name,
				cdr->party_a.snapshot->name);
			cdr_object_snapshot_copy(&cand_cdr->party_b, &cdr->party_a);
			cdr_all_relink(cand_cdr);
			/* It's possible that this joined at one point and was never chosen
			 * as party A. Clear their end time, as it would be set in such a
			 * case.
			 */
			memset(&cand_cdr->end, 0, sizeof(cand_cdr->end));
		}

		break;
	}

	ao2_unlock(base_cand_cdr);
}

/*!
 * \brief Handle creating bridge pairings for the \ref cdr_object that just
 * entered a bridge
 * \param cdr The \ref cdr_object that just entered the bridge
 * \param bridge The \ref ast_bridge_snapshot representing the bridge it just entered
 */
static void handle_bridge_pairings(struct cdr_object *cdr, struct ast_bridge_snapshot *bridge)
{
	struct ao2_iterator it_channels;
	char *channel_id;

	it_channels = ao2_iterator_init(bridge->channels, 0);
	while ((channel_id = ao2_iterator_next(&it_channels))) {
		struct cdr_object *cand_cdr;

		cand_cdr = ao2_find(active_cdrs_master, channel_id, OBJ_SEARCH_KEY);
		if (cand_cdr) {
			bridge_candidate_process(cdr, cand_cdr);
			ao2_ref(cand_cdr, -1);
		}

		ao2_ref(channel_id, -1);
	}
	ao2_iterator_destroy(&it_channels);
}

/*! \brief Handle entering into a parking bridge
 * \param cdr The CDR to operate on
 * \param bridge The bridge the channel just entered
 * \param channel The channel snapshot
 */
static void handle_parking_bridge_enter_message(struct cdr_object *cdr,
		struct ast_bridge_snapshot *bridge,
		struct ast_channel_snapshot *channel)
{
	int res = 1;
	struct cdr_object *it_cdr;
	struct cdr_object *new_cdr;

	ao2_lock(cdr);

	for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
		if (it_cdr->fn_table->process_parking_bridge_enter) {
			res &= it_cdr->fn_table->process_parking_bridge_enter(it_cdr, bridge, channel);
		}
		if (it_cdr->fn_table->process_party_a) {
			CDR_DEBUG("%p - Updating Party A %s snapshot\n", it_cdr,
				channel->name);
			it_cdr->fn_table->process_party_a(it_cdr, channel);
		}
	}

	if (res) {
		/* No one handled it - we need a new one! */
		new_cdr = cdr_object_create_and_append(cdr);
		if (new_cdr) {
			/* Let the single state transition us to Parked */
			cdr_object_transition_state(new_cdr, &single_state_fn_table);
			new_cdr->fn_table->process_parking_bridge_enter(new_cdr, bridge, channel);
		}
	}
	ao2_unlock(cdr);
}

/*! \brief Handle a bridge enter message for a 'normal' bridge
 * \param cdr The CDR to operate on
 * \param bridge The bridge the channel just entered
 * \param channel The channel snapshot
 */
static void handle_standard_bridge_enter_message(struct cdr_object *cdr,
		struct ast_bridge_snapshot *bridge,
		struct ast_channel_snapshot *channel)
{
	enum process_bridge_enter_results result;
	struct cdr_object *it_cdr;
	struct cdr_object *new_cdr;
	struct cdr_object *handled_cdr = NULL;

	ao2_lock(cdr);

try_again:
	for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
		if (it_cdr->fn_table->process_party_a) {
			CDR_DEBUG("%p - Updating Party A %s snapshot\n", it_cdr,
				channel->name);
			it_cdr->fn_table->process_party_a(it_cdr, channel);
		}

		/* Notify all states that they have entered a bridge */
		if (it_cdr->fn_table->process_bridge_enter) {
			CDR_DEBUG("%p - Processing bridge enter for %s\n", it_cdr,
				channel->name);
			result = it_cdr->fn_table->process_bridge_enter(it_cdr, bridge, channel);
			switch (result) {
			case BRIDGE_ENTER_ONLY_PARTY:
				/* Fall through */
			case BRIDGE_ENTER_OBTAINED_PARTY_B:
				if (!handled_cdr) {
					handled_cdr = it_cdr;
				}
				break;
			case BRIDGE_ENTER_NEED_CDR:
				/* Pass */
				break;
			case BRIDGE_ENTER_NO_PARTY_B:
				/* We didn't win on any - end this CDR. If someone else comes in later
				 * that is Party B to this CDR, it can re-activate this CDR.
				 */
				if (!handled_cdr) {
					handled_cdr = it_cdr;
				}
				cdr_object_finalize(cdr);
				break;
			}
		}
	}

	/* Create the new matchings, but only for either:
	 *  * The first CDR in the chain that handled it. This avoids issues with
	 *    forked CDRs.
	 *  * If no one handled it, the last CDR in the chain. This would occur if
	 *    a CDR joined a bridge and it wasn't Party A for anyone. We still need
	 *    to make pairings with everyone in the bridge.
	 */
	if (handled_cdr) {
		handle_bridge_pairings(handled_cdr, bridge);
	} else {
		/* Nothing handled it - we need a new one! */
		new_cdr = cdr_object_create_and_append(cdr);
		if (new_cdr) {
			/* This is guaranteed to succeed: the new CDR is created in the single state
			 * and will be able to handle the bridge enter message
			 */
			goto try_again;
		}
	}
	ao2_unlock(cdr);
}

/*!
 * \internal
 * \brief Handler for Stasis-Core bridge enter messages
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message The message - hopefully a bridge one!
 */
static void handle_bridge_enter_message(void *data, struct stasis_subscription *sub,
		struct stasis_message *message)
{
	struct ast_bridge_blob *update = stasis_message_data(message);
	struct ast_bridge_snapshot *bridge = update->bridge;
	struct ast_channel_snapshot *channel = update->channel;
	struct cdr_object *cdr;

	if (filter_bridge_messages(bridge)) {
		return;
	}

	if (filter_channel_snapshot(channel)) {
		return;
	}

	CDR_DEBUG("Bridge Enter message for channel %s: %u.%08u\n",
		channel->name,
		(unsigned int)stasis_message_timestamp(message)->tv_sec,
		(unsigned int)stasis_message_timestamp(message)->tv_usec);

	cdr = ao2_find(active_cdrs_master, channel->uniqueid, OBJ_SEARCH_KEY);
	if (!cdr) {
		ast_log(AST_LOG_WARNING, "No CDR for channel %s\n", channel->name);
		ast_assert(0);
		return;
	}

	if (!strcmp(bridge->subclass, "parking")) {
		handle_parking_bridge_enter_message(cdr, bridge, channel);
	} else {
		handle_standard_bridge_enter_message(cdr, bridge, channel);
	}
	ao2_cleanup(cdr);
}

/*!
 * \brief Handler for when a channel is parked
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message The message about who got parked
 * */
static void handle_parked_call_message(void *data, struct stasis_subscription *sub,
		struct stasis_message *message)
{
	struct ast_parked_call_payload *payload = stasis_message_data(message);
	struct ast_channel_snapshot *channel = payload->parkee;
	struct cdr_object *cdr;
	int unhandled = 1;
	struct cdr_object *it_cdr;

	/* Anything other than getting parked will be handled by other updates */
	if (payload->event_type != PARKED_CALL) {
		return;
	}

	/* No one got parked? */
	if (!channel) {
		return;
	}

	if (filter_channel_snapshot(channel)) {
		return;
	}

	CDR_DEBUG("Parked Call message for channel %s: %u.%08u\n",
		channel->name,
		(unsigned int)stasis_message_timestamp(message)->tv_sec,
		(unsigned int)stasis_message_timestamp(message)->tv_usec);

	cdr = ao2_find(active_cdrs_master, channel->uniqueid, OBJ_SEARCH_KEY);
	if (!cdr) {
		ast_log(AST_LOG_WARNING, "No CDR for channel %s\n", channel->name);
		ast_assert(0);
		return;
	}

	ao2_lock(cdr);

	for (it_cdr = cdr; it_cdr; it_cdr = it_cdr->next) {
		if (it_cdr->fn_table->process_parked_channel) {
			unhandled &= it_cdr->fn_table->process_parked_channel(it_cdr, payload);
		}
	}

	if (unhandled) {
		/* Nothing handled the messgae - we need a new one! */
		struct cdr_object *new_cdr;

		new_cdr = cdr_object_create_and_append(cdr);
		if (new_cdr) {
			/* As the new CDR is created in the single state, it is guaranteed
			 * to have a function for the parked call message and will handle
			 * the message */
			new_cdr->fn_table->process_parked_channel(new_cdr, payload);
		}
	}

	ao2_unlock(cdr);

	ao2_cleanup(cdr);
}

/*!
 * \brief Handler for a synchronization message
 * \param data Passed on
 * \param sub The stasis subscription for this message callback
 * \param topic The topic this message was published for
 * \param message A blank ao2 object
 * */
static void handle_cdr_sync_message(void *data, struct stasis_subscription *sub,
		struct stasis_message *message)
{
	return;
}

struct ast_cdr_config *ast_cdr_get_config(void)
{
	struct ast_cdr_config *general;
	struct module_config *mod_cfg;

	mod_cfg = ao2_global_obj_ref(module_configs);
	if (!mod_cfg) {
		return NULL;
	}
	general = ao2_bump(mod_cfg->general);
	ao2_cleanup(mod_cfg);
	return general;
}

void ast_cdr_set_config(struct ast_cdr_config *config)
{
	struct module_config *mod_cfg;

	if (!config) {
		return;
	}

	mod_cfg = ao2_global_obj_ref(module_configs);
	if (!mod_cfg) {
		return;
	}

	ao2_replace(mod_cfg->general, config);

	cdr_set_debug_mode(mod_cfg);
	cdr_toggle_runtime_options();

	ao2_cleanup(mod_cfg);
}

int ast_cdr_is_enabled(void)
{
	return is_cdr_flag_set(CDR_ENABLED);
}

int ast_cdr_backend_suspend(const char *name)
{
	int success = -1;
	struct cdr_beitem *i = NULL;

	AST_RWLIST_WRLOCK(&be_list);
	AST_RWLIST_TRAVERSE(&be_list, i, list) {
		if (!strcasecmp(name, i->name)) {
			ast_debug(3, "Suspending CDR backend %s\n", i->name);
			i->suspended = 1;
			success = 0;
		}
	}
	AST_RWLIST_UNLOCK(&be_list);

	return success;
}

int ast_cdr_backend_unsuspend(const char *name)
{
	int success = -1;
	struct cdr_beitem *i = NULL;

	AST_RWLIST_WRLOCK(&be_list);
	AST_RWLIST_TRAVERSE(&be_list, i, list) {
		if (!strcasecmp(name, i->name)) {
			ast_debug(3, "Unsuspending CDR backend %s\n", i->name);
			i->suspended = 0;
			success = 0;
		}
	}
	AST_RWLIST_UNLOCK(&be_list);

	return success;
}

int ast_cdr_register(const char *name, const char *desc, ast_cdrbe be)
{
	struct cdr_beitem *i = NULL;

	if (!name)
		return -1;

	if (!be) {
		ast_log(LOG_WARNING, "CDR engine '%s' lacks backend\n", name);
		return -1;
	}

	AST_RWLIST_WRLOCK(&be_list);
	AST_RWLIST_TRAVERSE(&be_list, i, list) {
		if (!strcasecmp(name, i->name)) {
			ast_log(LOG_WARNING, "Already have a CDR backend called '%s'\n", name);
			AST_RWLIST_UNLOCK(&be_list);
			return -1;
		}
	}

	if (!(i = ast_calloc(1, sizeof(*i))))
		return -1;

	i->be = be;
	ast_copy_string(i->name, name, sizeof(i->name));
	ast_copy_string(i->desc, desc, sizeof(i->desc));

	AST_RWLIST_INSERT_HEAD(&be_list, i, list);
	AST_RWLIST_UNLOCK(&be_list);

	return 0;
}

int ast_cdr_unregister(const char *name)
{
	struct cdr_beitem *match = NULL;
	int active_count;

	AST_RWLIST_WRLOCK(&be_list);
	AST_RWLIST_TRAVERSE(&be_list, match, list) {
		if (!strcasecmp(name, match->name)) {
			break;
		}
	}

	if (!match) {
		AST_RWLIST_UNLOCK(&be_list);
		return 0;
	}

	active_count = ao2_container_count(active_cdrs_master);

	if (!match->suspended && active_count != 0) {
		AST_RWLIST_UNLOCK(&be_list);
		ast_log(AST_LOG_WARNING, "Unable to unregister CDR backend %s; %d CDRs are still active\n",
			name, active_count);
		return -1;
	}

	AST_RWLIST_REMOVE(&be_list, match, list);
	AST_RWLIST_UNLOCK(&be_list);

	ast_verb(2, "Unregistered '%s' CDR backend\n", name);
	ast_free(match);

	return 0;
}

struct ast_cdr *ast_cdr_dup(struct ast_cdr *cdr)
{
	struct ast_cdr *newcdr;

	if (!cdr) {
		return NULL;
	}
	newcdr = ast_cdr_alloc();
	if (!newcdr) {
		return NULL;
	}

	*newcdr = *cdr;
	AST_LIST_HEAD_INIT_NOLOCK(&newcdr->varshead);
	copy_variables(&newcdr->varshead, &cdr->varshead);
	newcdr->next = NULL;

	return newcdr;
}

static const char *cdr_format_var_internal(struct ast_cdr *cdr, const char *name)
{
	struct ast_var_t *variables;

	if (ast_strlen_zero(name)) {
		return NULL;
	}

	AST_LIST_TRAVERSE(&cdr->varshead, variables, entries) {
		if (!strcasecmp(name, ast_var_name(variables))) {
			return ast_var_value(variables);
		}
	}

	return NULL;
}

static void cdr_get_tv(struct timeval when, const char *fmt, char *buf, int bufsize)
{
	if (fmt == NULL) {	/* raw mode */
		snprintf(buf, bufsize, "%ld.%06ld", (long)when.tv_sec, (long)when.tv_usec);
	} else {
		buf[0] = '\0';/* Ensure the buffer is initialized. */
		if (when.tv_sec) {
			struct ast_tm tm;

			ast_localtime(&when, &tm, NULL);
			ast_strftime(buf, bufsize, fmt, &tm);
		}
	}
}

void ast_cdr_format_var(struct ast_cdr *cdr, const char *name, char **ret, char *workspace, int workspacelen, int raw)
{
	const char *fmt = "%Y-%m-%d %T";
	const char *varbuf;

	if (!cdr) {
		return;
	}

	*ret = NULL;