Function handle_reply_jsonrpc

modules/event_stream/stream_send.c:485–587

}

static void handle_reply_jsonrpc(struct stream_con *con)
{
	/* got a reply on the connection */
	str buf;
	cJSON *reply;
	int bytes_read;
	const char *end;
	char buffer[STREAM_BUFFER_SIZE + 1];

	do {
		bytes_read = read(con->fd, buffer, STREAM_BUFFER_SIZE);
	} while (bytes_read == -1 && errno == EINTR);
	if (bytes_read < 0) {
		LM_ERR("error while reading reply from %s:%hu\n", STREAM_ADDR(con));
		goto error;
	} else if (bytes_read == 0) {
			LM_INFO("connection to %s:%hu closed!\n", STREAM_ADDR(con));
		goto error;
	}

	/* if not in reliable mode, we are not interested in the reply */
	if (stream_reliable_mode == 0)
		return;

	/* got a reply - parse it and match a command */
	/* TODO: proper parse a reply */
	LM_INFO("Received reply %.*s\n", bytes_read, buffer);

	/* if there was something else in the buffer, merge with what we had */
	if (con->pending_buffer.len) {
		/* XXX: this wasn't tested */
		con->pending_buffer.s = pkg_realloc(con->pending_buffer.s,
				con->pending_buffer.len + bytes_read + 1);
		if (!con->pending_buffer.s) {
			LM_ERR("No more pkg memory to keep replies!\n");
			goto error;
		}
		memcpy(con->pending_buffer.s + con->pending_buffer.len, buffer, bytes_read);
		con->pending_buffer.len += bytes_read;
		con->pending_buffer.s[con->pending_buffer.len] = 0;
		buf = con->pending_buffer;
	} else {
		buf.s = buffer;
		buf.len = bytes_read;
	}

	do {
		reply = cJSON_ParseWithOpts(buf.s, &end, 0);
		if (!reply && buf.s == end) {
			LM_ERR("cannot parse reply [%.*s]\n", buf.len, buf.s);
			goto error;
		}

		if (reply) {
			if (handle_cmd_reply(con, reply) < 0) {
				cJSON_Delete(reply);
				goto error;
			}