Function send_sync_repl

modules/clusterer/sync.c:333–408

}

void send_sync_repl(int sender, void *param)
{
	bin_packet_t sync_end_pkt, *pkt, *next_pkt;
	str bin_buffer;
	struct local_cap *cap;
	int rc, cluster_id, pkt_no = 0;
	struct reply_rpc_params *p = (struct reply_rpc_params *)param;

	for (cap = p->cluster->capabilities; cap; cap = cap->next)
		if (!str_strcmp(&p->cap_name, &cap->reg.name))
			break;
	if (!cap) {
		LM_ERR("Sync request for unknown capability: %.*s\n",
			p->cap_name.len, p->cap_name.s);
		goto out_free;
	}

	no_sync_chunks_sent = 0;

	cap->reg.event_cb(SYNC_REQ_RCV, p->node_id);

	lock_start_read(cl_list_lock);

	if (sync_packets) {
		bin_get_buffer(sync_packet_last, &bin_buffer);
		*sync_last_chunk_sz = bin_buffer.len - sync_prev_buf_len;

		/* send and free the lastly built packet */
		msg_add_trailer(sync_packet_last, p->cluster->cluster_id, p->node_id);

		for (pkt = sync_packets; pkt; pkt = next_pkt) {
			next_pkt = pkt->next;

			if ((rc = clusterer_send_msg(pkt, p->cluster->cluster_id,
				p->node_id, 0, 1))<0)
				LM_ERR("Failed to send sync packet, rc=%d\n", rc);

			bin_free_packet(pkt);
			free(pkt);
		}

		sync_packets = NULL;
		pkt_no = sync_packets_cnt;
		sync_packets_cnt = 0;
		sync_packet_last = NULL;
		sync_last_chunk_sz = NULL;
	}

	/* send indication that all sync packets were sent */
	if (bin_init(&sync_end_pkt,&cl_extra_cap,CLUSTERER_SYNC_END,BIN_SYNC_VERSION,0)<0) {
		LM_ERR("Failed to init bin packet\n");
		lock_stop_read(cl_list_lock);
		goto out_free;
	}
	bin_push_str(&sync_end_pkt, &p->cap_name);
	bin_push_int(&sync_end_pkt, no_sync_chunks_sent);
	msg_add_trailer(&sync_end_pkt, p->cluster->cluster_id, p->node_id);