Function handle_sync_packet

modules/clusterer/sync.c:564–643

}

void handle_sync_packet(bin_packet_t *packet, int packet_type,
								cluster_info_t *cluster, int source_id)
{
	str cap_name;
	struct local_cap *cap;
	int data_version;
	int no_sync_chunks;
	int was_in_progress = 0;

	if (get_bin_pkg_version(packet) != BIN_SYNC_VERSION) {
		LM_INFO("discarding sync packet version %d, need version %d\n",
		        get_bin_pkg_version(packet), BIN_SYNC_VERSION);
		return;
	}

	bin_pop_str(packet, &cap_name);
	for (cap = cluster->capabilities; cap; cap = cap->next)
		if (!str_strcmp(&cap_name, &cap->reg.name))
			break;
	if (!cap) {
		LM_ERR("Capability: %.*s from sync packet, not found\n",
			cap_name.len, cap_name.s);
		return;
	}

	if (get_capability_status(cluster, &cap_name) != 1) {
		LM_DBG("capability disabled, drop sync packet\n");
		return;
	}

	if (packet_type == CLUSTERER_SYNC) {
		bin_pop_int(packet, &data_version);

		lock_get(cluster->lock);

		/* if the cap's state is already OK (e.g. donor aborted sync mid-way,
		 * then sync_check_timer() timed out the sync back to CAP_STATE_OK),
		 * avoid forcing a state where repl packets queue indefinitely! */
		if (!(cap->flags & CAP_STATE_OK)) {
			if (cap->flags & CAP_SYNC_IN_PROGRESS)
				was_in_progress = 1;

			/* buffer other types of packets during sync */
			cap->flags |= CAP_SYNC_IN_PROGRESS;
		}

		cap->last_sync_pkt = get_ticks();
		lock_release(cluster->lock);

		if (!was_in_progress) {
			sr_set_status(cl_srg, STR2CI(cap->reg.sr_id), CAP_SR_SYNCING,
				STR2CI(CAP_SR_STATUS_STR(CAP_SR_SYNCING)), 0);
			sr_add_report_fmt(cl_srg, STR2CI(cap->reg.sr_id), 0,
				"Sync started from node [%d]", source_id);
		}

		/* overwrite packet type with one identifiable by modules */
		packet->type = SYNC_PACKET_TYPE;