rtpp_command.c

大名鼎鼎的 RTP Proxy 源代码 在OpenSER 中成熟应用的

    }
	
    if (op == DELETE || op == RECORD || op == COPY || op == NOPLAY) {
		/* D, R and S commands don't take any modifiers */
		if (argv[0][1] != '\0') {
		    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
		    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
		    return 0;
		}
    }
	
    if (op == UPDATE || op == LOOKUP || op == DELETE) {
		addr = argv[2];
		port = argv[3];
		/* Process additional command modifiers */
		external = 1;
		/* In bridge mode all clients are assumed to be asymmetric */
		asymmetric = (cf->bmode != 0) ? 1 : 0;
		pf = AF_INET;
		weak = 0;
		for (cp = argv[0] + 1; *cp != '\0'; cp++) {
		    switch (*cp) {
		    case 'a':
		    case 'A':
			asymmetric = 1;
			break;

		    case 'e':
		    case 'E':
			if (lidx < 0) {
			    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
			    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
			    return 0;
			}
			lia[lidx] = cf->bindaddr[1];
			lidx--;
			break;

		    case 'i':
		    case 'I':
			if (lidx < 0) {
			    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
			    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
			    return 0;
			}
			lia[lidx] = cf->bindaddr[0];
			lidx--;
			break;

		    case '6':
			pf = AF_INET6;
			break;

		    case 's':
		    case 'S':
			asymmetric = 0;
			break;

		    case 'w':
		    case 'W':
			weak = 1;
			break;

		    case 'z':
		    case 'Z':
			requested_nsamples = (strtol(cp + 1, &cp, 10) / 10) * 80;
			if (requested_nsamples <= 0) {
			    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "command syntax error");
			    reply_error(cf, controlfd, &raddr, rlen, cookie, 1);
			    return 0;
			}
			cp--;
			break;

		    default:
			rtpp_log_write(RTPP_LOG_ERR, cf->glog, "unknown command modifier `%c'",
			  *cp);
			break;
		    }
		}
		
		if (op != DELETE && addr != NULL && port != NULL && strlen(addr) >= 7) {
		    struct sockaddr_storage tia;

		    if ((n = resolve(sstosa(&tia), pf, addr, port,
		      AI_NUMERICHOST)) == 0) {
			if (!ishostnull(sstosa(&tia))) {
			    for (i = 0; i < 2; i++) {
				ia[i] = malloc(SS_LEN(&tia));
				if (ia[i] == NULL) {
				    handle_nomem(cf, controlfd, &raddr, rlen, cookie,
				      5, ia, fds, spa, spb);
				    return 0;
				}
				memcpy(ia[i], &tia, SS_LEN(&tia));
			    }
			    /* Set port for RTCP, will work both for IPv4 and IPv6 */
			    n = ntohs(satosin(ia[1])->sin_port);
			    satosin(ia[1])->sin_port = htons(n + 1);
			}
		    } else {
			rtpp_log_write(RTPP_LOG_ERR, cf->glog, "getaddrinfo: %s",
			  gai_strerror(n));
		    }
		}
    }

    if (op != DELETE) {
		i = find_stream(cf, call_id, from_tag, to_tag, &spa);
		if (i != -1 && op != UPDATE)
		    i = NOT(i);
	    } else {
			i = handle_delete(cf, call_id, from_tag, to_tag, weak);
    	}

	    if (i == -1 && op != UPDATE) {
			rtpp_log_write(RTPP_LOG_INFO, cf->glog,
			  "%s request failed: session %s, tags %s/%s not found", rname,
			  call_id, from_tag, to_tag != NULL ? to_tag : "NONE");
				if (op == LOOKUP) {
				    for (i = 0; i < 2; i++)
					if (ia[i] != NULL)
					    free(ia[i]);
				    reply_port(cf, controlfd, &raddr, rlen, cookie, 0, lia);
				    return 0;
				}
			
			reply_error(cf, controlfd, &raddr, rlen, cookie, 8);
			return 0;
	    }

	    switch (op) {
		    case DELETE:
				reply_ok(cf, controlfd, &raddr, rlen, cookie);
				return 0;

		    case NOPLAY:
				handle_noplay(cf, spa, i);
				reply_ok(cf, controlfd, &raddr, rlen, cookie);
				return 0;

		    case PLAY:
				handle_noplay(cf, spa, i);
				if (playcount != 0 && handle_play(cf, spa, i, codecs, pname, playcount) != 0) {
				    reply_error(cf, controlfd, &raddr, rlen, cookie, 6);
				    return 0;
				}
				reply_ok(cf, controlfd, &raddr, rlen, cookie);
				return 0;

		    case COPY:
		    case RECORD:
				handle_record(cf, spa, i, recording_name);
				reply_ok(cf, controlfd, &raddr, rlen, cookie);
				return 0;

		    case QUERY:
				handle_query(cf, controlfd, &raddr, rlen, cookie, spa, i);
				return 0;
	    }

    pidx = 1;
    lport = 0;
    if (i != -1) {
		assert(op == UPDATE || op == LOOKUP);
		if (spa->fds[i] == -1) {
		    j = ishostseq(cf->bindaddr[0], spa->laddr[i]) ? 0 : 1;
		    if (create_listener(cf, spa->laddr[i], cf->nextport[j],
			      &lport, fds) == -1) {
				rtpp_log_write(RTPP_LOG_ERR, spa->log, "can't create listener");
				reply_error(cf, controlfd, &raddr, rlen, cookie, 7);
				return 0;
		    }
		    cf->nextport[j] = lport + 2;
		    assert(spa->fds[i] == -1);
		    spa->fds[i] = fds[0];
		    assert(spa->rtcp->fds[i] == -1);
		    spa->rtcp->fds[i] = fds[1];
		    spa->ports[i] = lport;
		    spa->rtcp->ports[i] = lport + 1;
		    spa->complete = spa->rtcp->complete = 1;
		    append_session(cf, spa, i);
		    append_session(cf, spa->rtcp, i);
		}
	
	if (weak)
	    spa->weak[i] = 1;
	else if (op == UPDATE)
	    spa->strong = 1;
	
	lport = spa->ports[i];
	lia[0] = spa->laddr[i];
	pidx = (i == 0) ? 1 : 0;
	spa->ttl = cf->max_ttl;
	
	if (op == UPDATE) {
	    rtpp_log_write(RTPP_LOG_INFO, spa->log,
	      "adding %s flag to existing session, new=%d/%d/%d",
	      weak ? ( i ? "weak[1]" : "weak[0]" ) : "strong",
	      spa->strong, spa->weak[0], spa->weak[1]);
	}
	
	rtpp_log_write(RTPP_LOG_INFO, spa->log,
	  "lookup on ports %d/%d, session timer restarted", spa->ports[0],
	  spa->ports[1]);
    } else {
	assert(op == UPDATE);
	rtpp_log_write(RTPP_LOG_INFO, cf->glog,
	  "new session %s, tag %s requested, type %s",
	  call_id, from_tag, weak ? "weak" : "strong");

	j = ishostseq(cf->bindaddr[0], lia[0]) ? 0 : 1;
	if (create_listener(cf, cf->bindaddr[j], cf->nextport[j], &lport,
	  fds) == -1) {
	    rtpp_log_write(RTPP_LOG_ERR, cf->glog, "can't create listener");
	    reply_error(cf, controlfd, &raddr, rlen, cookie, 10);
	    return 0;
	}
	cf->nextport[j] = lport + 2;

	/*
	 * Session creation. If creation is requested with weak flag,
	 * set weak[0].
	 */
	spa = malloc(sizeof(*spa));
	if (spa == NULL) {
	    handle_nomem(cf, controlfd, &raddr, rlen, cookie, 11, ia,
	      fds, spa, spb);
	    return 0;
	}
	/* spb is RTCP twin session for this one. */
	spb = malloc(sizeof(*spb));
	if (spb == NULL) {
	    handle_nomem(cf, controlfd, &raddr, rlen, cookie, 12, ia,
	      fds, spa, spb);
	    return 0;
	}
	memset(spa, 0, sizeof(*spa));
	memset(spb, 0, sizeof(*spb));
	for (i = 0; i < 2; i++)
	    spa->fds[i] = spb->fds[i] = -1;
	spa->call_id = strdup(call_id);
	if (spa->call_id == NULL) {
	    handle_nomem(cf, controlfd, &raddr, rlen, cookie, 13, ia,
	      fds, spa, spb);
	    return 0;
	}
	spb->call_id = spa->call_id;
	spa->tag = strdup(from_tag);
	if (spa->tag == NULL) {
	    handle_nomem(cf, controlfd, &raddr, rlen, cookie, 14, ia,
	      fds, spa, spb);
	    return 0;
	}
	spb->tag = spa->tag;
	for (i = 0; i < 2; i++) {
	    spa->rrcs[i] = NULL;
	    spb->rrcs[i] = NULL;
	    spa->laddr[i] = lia[i];
	    spb->laddr[i] = lia[i];
	}
	spa->strong = spa->weak[0] = spa->weak[1] = 0;
	if (weak)
	    spa->weak[0] = 1;
	else
	    spa->strong = 1;
	assert(spa->fds[0] == -1);
	spa->fds[0] = fds[0];
	assert(spb->fds[0] == -1);
	spb->fds[0] = fds[1];
	spa->ports[0] = lport;
	spb->ports[0] = lport + 1;
	spa->ttl = cf->max_ttl;
	spb->ttl = -1;
	spa->log = rtpp_log_open("rtpproxy", spa->call_id, 0);
	spb->log = spa->log;
	spa->rtcp = spb;
	spb->rtcp = NULL;
	spa->rtp = NULL;
	spb->rtp = spa;
	spa->sridx = spb->sridx = -1;

	append_session(cf, spa, 0);
	append_session(cf, spa, 1);
	append_session(cf, spb, 0);
	append_session(cf, spb, 1);

	hash_table_append(cf, spa);

	cf->sessions_created++;
	cf->sessions_active++;
	/*
	 * Each session can consume up to 5 open file descriptors (2 RTP,
	 * 2 RTCP and 1 logging) so that warn user when he is likely to
	 * exceed 80% mark on hard limit.
	 */
	if (cf->sessions_active > (cf->nofile_limit.rlim_max * 80 / (100 * 5)) &&
	  cf->nofile_limit_warned == 0) {
	    cf->nofile_limit_warned = 1;
	    rtpp_log_write(RTPP_LOG_WARN, cf->glog, "passed 80%% "
	      "threshold on the open file descriptors limit (%d), "
	      "consider increasing the limit using -L command line "
	      "option", (int)cf->nofile_limit.rlim_max);
	}

	rtpp_log_write(RTPP_LOG_INFO, spa->log, "new session on a port %d created, "
	  "tag %s", lport, from_tag);
    }

    if (ia[0] != NULL && ia[1] != NULL) {
	/*
	 * Unless the address provided by client historically
	 * cannot be trusted and address is different from one
	 * that we recorded update it.
	 */
	if (spa->untrusted_addr[pidx] == 0 && !(spa->addr[pidx] != NULL &&
	  SA_LEN(ia[0]) == SA_LEN(spa->addr[pidx]) &&
	  memcmp(ia[0], spa->addr[pidx], SA_LEN(ia[0])) == 0)) {
	    rtpp_log_write(RTPP_LOG_INFO, spa->log, "pre-filling %s's address "
	      "with %s:%s", (pidx == 0) ? "callee" : "caller", addr, port);
	    if (spa->addr[pidx] != NULL)
		free(spa->addr[pidx]);
	    spa->addr[pidx] = ia[0];
	    ia[0] = NULL;
	}
	if (spa->rtcp->untrusted_addr[pidx] == 0 && !(spa->rtcp->addr[pidx] != NULL &&
	  SA_LEN(ia[1]) == SA_LEN(spa->rtcp->addr[pidx]) &&
	  memcmp(ia[1], spa->rtcp->addr[pidx], SA_LEN(ia[1])) == 0)) {
	    if (spa->rtcp->addr[pidx] != NULL)
		free(spa->rtcp->addr[pidx]);
	    spa->rtcp->addr[pidx] = ia[1];
	    ia[1] = NULL;
	}
    }
    spa->asymmetric[pidx] = spa->rtcp->asymmetric[pidx] = asymmetric;
    spa->canupdate[pidx] = spa->rtcp->canupdate[pidx] = NOT(asymmetric);
    if (requested_nsamples > 0) {
	rtpp_log_write(RTPP_LOG_INFO, spa->log, "RTP packets from %s "
	  "will be resized to %d milliseconds",
	  (pidx == 0) ? "callee" : "caller", requested_nsamples / 8);
    } else if (spa->resizers[pidx].output_nsamples > 0) {
	  rtpp_log_write(RTPP_LOG_INFO, spa->log, "Resizing of RTP "
	  "packets from %s has been disabled",
	  (pidx == 0) ? "callee" : "caller");
    }
    spa->resizers[pidx].output_nsamples = requested_nsamples;

    for (i = 0; i < 2; i++)
	if (ia[i] != NULL)
	    free(ia[i]);

    assert(lport != 0);
    reply_port(cf, controlfd, &raddr, rlen, cookie, lport, lia);
    return 0;
}

static int
handle_delete(struct cfg *cf, char *call_id, char *from_tag, char *to_tag, int weak)
{
    int medianum, ndeleted;
    struct rtpp_session *spa, *spb;
    int cmpr, cmpr1, idx;

    ndeleted = 0;
    for (spa = hash_table_findfirst(cf, call_id); spa != NULL;) {
	medianum = 0;
	if ((cmpr1 = compare_session_tags(spa->tag, from_tag, &medianum)) != 0) {
	    idx = 1;
	    cmpr = cmpr1;
	} else if (to_tag != NULL &&
	  (cmpr1 = compare_session_tags(spa->tag, to_tag, &medianum)) != 0) {
	    idx = 0;
	    cmpr = cmpr1;
	} else {
	    spa = hash_table_findnext(spa);
	    continue;
	}

	if (weak)
	    spa->weak[idx] = 0;
	else
	    spa->strong = 0;

	/*
	 * This seems to be stable from reiterations, the only side
	 * effect is less efficient work.
	 */
	if (spa->strong || spa->weak[0] || spa->weak[1]) {
	    rtpp_log_write(RTPP_LOG_INFO, spa->log,
	      "delete: medianum=%u: removing %s flag, seeing flags to"
	      " continue session (strong=%d, weak=%d/%d)",
	      medianum,
	      weak ? ( idx ? "weak[1]" : "weak[0]" ) : "strong",
	      spa->strong, spa->weak[0], spa->weak[1]);
	    /* Skipping to next possible stream for this call */
	    ++ndeleted;
	    spa = hash_table_findnext(spa);
	    continue;
	}
	rtpp_log_write(RTPP_LOG_INFO, spa->log,
	  "forcefully deleting session %d on ports %d/%d",
	   medianum, spa->ports[0], spa->ports[1]);
	/* Search forward before we do removal */
	spb = spa;
	spa = hash_table_findnext(spa);
	remove_session(cf, spb);
	++ndeleted;
	if (cmpr != 2) {
	    break;
	}
    }
    if (ndeleted == 0) {
	return -1;
    }
    return 0;
}

static void
handle_noplay(struct cfg *cf, struct rtpp_session *spa, int idx)
{

    if (spa->rtps[idx] != NULL) {
	rtp_server_free(spa->rtps[idx]);
	spa->rtps[idx] = NULL;
	rtpp_log_write(RTPP_LOG_INFO, spa->log,
	  "stopping player at port %d", spa->ports[idx]);
	if (spa->rtps[0] == NULL && spa->rtps[1] == NULL) {
	    assert(cf->rtp_servers[spa->sridx] == spa);
	    cf->rtp_servers[spa->sridx] = NULL;
	    spa->sridx = -1;
	}
   }
}

static int
handle_play(struct cfg *cf, struct rtpp_session *spa, int idx, char *codecs,
  char *pname, int playcount)
{
    int n;
    char *cp;

    while (*codecs != '\0') {
		n = strtol(codecs, &cp, 10);
		if (cp == codecs)
		    break;
		codecs = cp;
		if (*codecs != '\0')
		    codecs++;
		
		spa->rtps[idx] = rtp_server_new(pname, n, playcount);
		
		if (spa->rtps[idx] == NULL)
		    continue;
		
		rtpp_log_write(RTPP_LOG_INFO, spa->log,
		  "%d times playing prompt %s codec %d", playcount, pname, n);
		
		if (spa->sridx == -1)
		    append_server(cf, spa);
		
		return 0;
    }
	
    rtpp_log_write(RTPP_LOG_ERR, spa->log, "can't create player");
    return -1;
}

static void
handle_record(struct cfg *cf, struct rtpp_session *spa, int idx, char *rname)
{

    if (spa->rrcs[idx] == NULL) {
	spa->rrcs[idx] = ropen(cf, spa, rname, idx);
	rtpp_log_write(RTPP_LOG_INFO, spa->log,
	  "starting recording RTP session on port %d", spa->ports[idx]);
    }
    if (spa->rtcp->rrcs[idx] == NULL && cf->rrtcp != 0) {
	spa->rtcp->rrcs[idx] = ropen(cf, spa->rtcp, rname, idx);
	rtpp_log_write(RTPP_LOG_INFO, spa->log,
	  "starting recording RTCP session on port %d", spa->rtcp->ports[idx]);
    }
}

static void
handle_query(struct cfg *cf, int fd, struct sockaddr_storage *raddr,
  socklen_t rlen, char *cookie, struct rtpp_session *spa, int idx)
{
    char buf[1024 * 8];
    int len;

    if (cookie != NULL) {
	len = sprintf(buf, "%s %d %lu %lu %lu %lu\n", cookie, spa->ttl,
	  spa->pcount[idx], spa->pcount[NOT(idx)], spa->pcount[2],
	  spa->pcount[3]);
    } else {
	len = sprintf(buf, "%d %lu %lu %lu %lu\n", spa->ttl,
	  spa->pcount[idx], spa->pcount[NOT(idx)], spa->pcount[2],
	  spa->pcount[3]);
    }
    doreply(cf, fd, buf, len, raddr, rlen);
}