main_reading.c

ser作为一个高效、高度可配置的simple服务器

/*
 * Copyright (c) 2003 - 2005 Maxim Sobolev
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * $Id: main.c,v 1.31 2005/12/12 23:25:17 sobomax Exp $
 *
 */

#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/un.h>
#include <sys/uio.h>
#if defined(__FreeBSD__)
#include <sys/queue.h>
#else
#include "myqueue.h"
#endif
#include <sys/select.h>
#include <sys/stat.h>
#if !defined(__solaris__)
#include <err.h>
#endif
#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netdb.h>
#include <poll.h>
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <string.h>
#include <strings.h>
#include <unistd.h>

#include "rtp_server.h"
#include "rtpp_defines.h"
#include "rtpp_log.h"
#include "rtpp_record.h"
#include "rtpp_session.h"
#include "rtpp_util.h"

#define	GET_RTP(sp)	(((sp)->rtp != NULL) ? (sp)->rtp : (sp))
#define	NOT(x)		(((x) == 0) ? 1 : 0)

static LIST_HEAD(, rtpp_session) session_set = LIST_HEAD_INITIALIZER(&session_set);

static struct rtpp_session *sessions[MAX_FDS];
static struct rtpp_session *rtp_servers[MAX_FDS];
static struct pollfd fds[MAX_FDS + 1];
static int nsessions;
static unsigned long long sessions_created = 0;
static int rtp_nsessions;
static int bmode = 0;			/* Bridge mode */
static int umode = 0;			/* UDP control mode */
static const char *cmd_sock = CMD_SOCK;
static const char *pid_file = PID_FILE;

struct proto_cap {
    const char	*pc_id;
    const char	*pc_description;
};

static struct proto_cap proto_caps[] = {
    /*
     * The first entry must be basic protocol version and isn't shown
     * as extension on -v.
     */
    { "20040107", "Basic RTP proxy functionality" },
    { "20050322", "Support for multiple RTP streams and MOH" },
    { NULL, NULL }
};

/*
 * The first address is for external interface, the second one - for
 * internal one. Second can be NULL, in this case there is no bridge
 * mode enabled.
 */
static struct sockaddr *bindaddr[2];	/* RTP socket(s) addresses */

static rtpp_log_t glog;
static int tos = TOS;
static int lastport[2] = {PORT_MIN - 1, PORT_MIN - 1};
static const char *rdir = NULL;
static const char *sdir = NULL;
static int rrtcp = 1;

static void setbindhost(struct sockaddr *, int, const char *, const char *);
static void remove_session(struct rtpp_session *, struct rtpp_session **);
static void rebuild_tables(void);
static void alarmhandler(int);
static int create_twinlistener(struct sockaddr *, int, int *);
static int create_listener(struct sockaddr *, int, int, int, int *, int *);
static void handle_command(int);
static void usage(void);

static void
setbindhost(struct sockaddr *ia, int pf, const char *bindhost,
  const char *servname)
{
    int n;

    /*
     * If user specified * then change it to NULL,
     * that will make getaddrinfo to return addr_any socket
     */
    if (bindhost && (strcmp(bindhost, "*") == 0))//strcmp()比较bindhost和"*"是否相等，如果相等，bindhost设为空
	bindhost = NULL;

    if ((n = resolve(ia, pf, bindhost, servname, AI_PASSIVE)) != 0)//查找地址信息
	errx(1, "setbindhost: %s", gai_strerror(n));//写出错信息
}

static void
rebuild_tables(void) //填充
{
    struct rtpp_session *sp;
    int j;

    nsessions = 0;
    rtp_nsessions = 0;
    LIST_FOREACH(sp, &session_set, link) {
	for (j = 0; j < 2; j++) {
	    if (sp->fds[j] != -1) {
		sessions[nsessions] = sp;
		nsessions++;
		fds[nsessions].fd = sp->fds[j];
		fds[nsessions].events = POLLIN;
		fds[nsessions].revents = 0;
	    }
	}
	if (sp->rtps[0] != NULL || sp->rtps[1] != NULL) {
	    rtp_servers[rtp_nsessions] = sp;
	    rtp_nsessions++;
	}
    }
}

static void
alarmhandler(int sig __attribute__ ((unused)))
{
    struct rtpp_session *sp, *rsp;
    int changed;

    changed = 0;
    for(sp = LIST_FIRST(&session_set); sp != NULL; sp = rsp) {
	rsp = LIST_NEXT(sp, link);
	if (sp->rtcp == NULL)
	    continue;
	if (sp->ttl == 0) {
	    rtpp_log_write(RTPP_LOG_INFO, sp->log, "session timeout");
	    remove_session(sp, NULL);
	    changed = 1;
	    continue;
	}
	sp->ttl--;
    }
    if (changed == 1)
	rebuild_tables();
}

static void
remove_session(struct rtpp_session *sp, struct rtpp_session **pspiter)
{
    int i;
//pcount[]数组
    rtpp_log_write(RTPP_LOG_INFO, sp->log, "RTP stats: %lu in from callee, %lu "
      "in from caller, %lu relayed, %lu dropped", sp->pcount[0], sp->pcount[1],
      sp->pcount[2], sp->pcount[3]);
    rtpp_log_write(RTPP_LOG_INFO, sp->log, "RTCP stats: %lu in from callee, %lu "
      "in from caller, %lu relayed, %lu dropped", sp->rtcp->pcount[0],
      sp->rtcp->pcount[1], sp->rtcp->pcount[2], sp->rtcp->pcount[3]);
    rtpp_log_write(RTPP_LOG_INFO, sp->log, "session on ports %d/%d is cleaned up",
      sp->ports[0], sp->ports[1]);//port[]数组
    for (i = 0; i < 2; i++) {
	if (sp->addr[i] != NULL)
	    free(sp->addr[i]);
	if (sp->rtcp->addr[i] != NULL)
	    free(sp->rtcp->addr[i]);
	if (sp->fds[i] != -1)
	    close(sp->fds[i]);
	if (sp->rtcp->fds[i] != -1)
	    close(sp->rtcp->fds[i]);
	if (sp->rrcs[i] != NULL)
	    rclose(sp, sp->rrcs[i]);
	if (sp->rtcp->rrcs[i] != NULL)
	    rclose(sp, sp->rtcp->rrcs[i]);
	if (sp->rtps[i] != NULL)
	    rtp_server_free(sp->rtps[i]);
    }
    if (sp->call_id != NULL)
	free(sp->call_id);
    if (sp->tag != NULL)
	free(sp->tag);
    rtpp_log_close(sp->log);
    if (pspiter != NULL && *pspiter == sp)
	*pspiter = LIST_NEXT(sp, link);
    LIST_REMOVE(sp, link);
    if (pspiter != NULL && *pspiter == sp->rtcp)
	*pspiter = LIST_NEXT(sp->rtcp, link);
    LIST_REMOVE(sp->rtcp, link);
    free(sp->rtcp);
    free(sp);
}

static int
create_twinlistener(struct sockaddr *ia, int port, int *fds)
{
    struct sockaddr_storage iac;
    int rval, i, flags;

    fds[0] = fds[1] = -1;

    rval = -1;
    for (i = 0; i < 2; i++) {
	fds[i] = socket(ia->sa_family, SOCK_DGRAM, 0);
	if (fds[i] == -1) {
	    rtpp_log_ewrite(RTPP_LOG_ERR, glog, "can't create %s socket",
	      (ia->sa_family == AF_INET) ? "IPv4" : "IPv6");
	    goto failure;
	}
	memcpy(&iac, ia, SA_LEN(ia));
	satosin(&iac)->sin_port = htons(port);
	if (bind(fds[i], sstosa(&iac), SA_LEN(ia)) != 0) {
	    if (errno != EADDRINUSE && errno != EACCES) {
		rtpp_log_ewrite(RTPP_LOG_ERR, glog, "can't bind to the %s port %d",
		  (ia->sa_family == AF_INET) ? "IPv4" : "IPv6", port);
	    } else {
		rval = -2;
	    }
	    goto failure;
	}
	port++;
	if ((ia->sa_family == AF_INET) &&
	  (setsockopt(fds[i], IPPROTO_IP, IP_TOS, &tos, sizeof(tos)) == -1))
	    rtpp_log_ewrite(RTPP_LOG_ERR, glog, "unable to set TOS to %d", tos);
	flags = fcntl(fds[i], F_GETFL);
	fcntl(fds[i], F_SETFL, flags | O_NONBLOCK);
    }
    return 0;

failure:
    for (i = 0; i < 2; i++)
	if (fds[i] != -1) {
	    close(fds[i]);
	    fds[i] = -1;
	}
    return rval;
}

static int
create_listener(struct sockaddr *ia, int minport, int maxport,
  int startport, int *port, int *fds)
{
    int i, init, rval;

    /* make sure that {min,max,start}port is even */
    if ((minport & 0x1) != 0)
	minport++;
    if ((maxport & 0x1) != 0)
	maxport--;
    if ((startport & 0x1) != 0)
	startport++;

    for (i = 0; i < 2; i++)
	fds[i] = -1;

    init = 0;
    if (startport < minport || startport > maxport)
	startport = minport;
    for (*port = startport; *port != startport || init == 0; (*port) += 2) {
	init = 1;
	rval = create_twinlistener(ia, *port, fds);
	if (rval != 0) {
	    if (rval == -1)
		break;
	    if (*port >= maxport)
		*port = minport - 2;
	    continue;
	}
	return 0;
    }
    return -1;
}

static int
compare_session_tags(char *tag1, char *tag0, unsigned *medianum_p)
{
    size_t len0 = strlen(tag0);
    if (!strncmp(tag1, tag0, len0)) {
	if (tag1[len0] == ';') {
		if (medianum_p != 0)
			*medianum_p = strtoul(tag1+len0+1, NULL, 10);
		return 2;
	}
	if (tag1[len0] == 0) return 1;
	return 0;
    }
    return 0;
}

static void
handle_command(int controlfd)
{
    int len, delete, argc, i, j, pidx, request, response, asymmetric;
    int external, rlen, pf, ecode, lidx, play, record, noplay, weak;
    int ndeleted, skipnext, cmpr, cmpr1;
    int fds[2], lport, n;
    unsigned medianum;
    char buf[1024 * 8];
    char *cp, *call_id, *from_tag, *to_tag, *addr, *port, *cookie;
    char *pname, *codecs;
    struct rtpp_session *spa, *spb, *spnext;
    char **ap, *argv[10];
    const char *rname;
    struct sockaddr *ia[2], *lia[2];
    struct sockaddr_storage raddr;

#define	doreply() \
    { \
	buf[len] = '\0'; \
	rtpp_log_write(RTPP_LOG_DBUG, glog, "sending reply \"%s\"", buf); \
	if (umode == 0) { \
	    write(controlfd, buf, len); \
	} else { \
	    while (sendto(controlfd, buf, len, 0, sstosa(&raddr), \
	      rlen) == -1 && errno == ENOBUFS); \
	} \
    }

    ia[0] = ia[1] = NULL;
    spa = spb = NULL;
    lia[0] = lia[1] = bindaddr[0];
    lidx = 1;
    fds[0] = fds[1] = -1;

    if (umode == 0) {
	len = read(controlfd, buf, sizeof(buf) - 1);
    } else {
	rlen = sizeof(raddr);
	len = recvfrom(controlfd, buf, sizeof(buf) - 1, 0,
	  sstosa(&raddr), &rlen);
    }
    if (len == -1) {
	rtpp_log_ewrite(RTPP_LOG_ERR, glog, "can't read from control socket");
	return;
    }
    buf[len] = '\0';

    rtpp_log_write(RTPP_LOG_DBUG, glog, "received command \"%s\"", buf);

    cp = buf;
    argc = 0;
    memset(argv, 0, sizeof(argv));
    for (ap = argv; (*ap = strsep(&cp, "\r\n\t ")) != NULL;)
	if (**ap != '\0') {
	    argc++;
	    if (++ap >= &argv[10])
		break;
	}
    cookie = NULL;
    if (argc < 1 || (umode != 0 && argc < 2)) {
	rtpp_log_write(RTPP_LOG_ERR, glog, "command syntax error");
	ecode = 0;
	goto goterror;
    }

    /* Stream communication mode doesn't use cookie */
    if (umode != 0) {
	cookie = argv[0];
	for (i = 1; i < argc; i++)
	    argv[i - 1] = argv[i];
	argc--;
	argv[argc] = NULL;
    } else {
	cookie = NULL;
    }

    request = response = delete = play = record = noplay = 0;
    addr = port = NULL;
    switch (argv[0][0]) {
    case 'u':
    case 'U':
	request = 1;
	break;

    case 'l':
    case 'L':
	response = 1;
	break;

    case 'd':
    case 'D':
	delete = 1;
	break;

    case 'p':
    case 'P':
	/* P callid pname codecs from_tag to_tag */
	play = 1;
	pname = argv[2];
	codecs = argv[3];
	break;

    case 'r':
    case 'R':
        record = 1;
        break;

    case 's':
    case 'S':
        noplay = 1;
        break;

    case 'v':
    case 'V':
	if (argv[0][1] == 'F' || argv[0][1] == 'f') {
	    int i, known;
	    /*
	     * Wait for protocol version datestamp and check whether we
	     * know it.
	     */
	    if (argc != 2) {
		rtpp_log_write(RTPP_LOG_ERR, glog, "command syntax error");
		ecode = 2;
		goto goterror;
	    }
	    for (known = i = 0; proto_caps[i].pc_id != NULL; ++i) {
		if (!strcmp(argv[1], proto_caps[i].pc_id)) {
		    known = 1;
		    break;
		}
	    }
	    if (cookie == NULL)
		len = sprintf(buf, "%d\n", known);
	    else
		len = sprintf(buf, "%s %d\n", cookie, known);
	    goto doreply;
	}
	if (argc != 1) {
	    rtpp_log_write(RTPP_LOG_ERR, glog, "command syntax error");
	    ecode = 2;
	    goto goterror;
	}
	/* This returns base version. */
	if (cookie == NULL)
	    len = sprintf(buf, "%d\n", CPROTOVER);
	else
	    len = sprintf(buf, "%s %d\n", cookie, CPROTOVER);
	goto doreply;
	break;

    case 'i':
    case 'I':
	len = sprintf(buf, "sessions created: %llu\nactive sessions:\n", sessions_created);
	LIST_FOREACH(spa, &session_set, link) {