录制UDP组播数据到文件(1)

简介:
网上的通常的UDP组播程序都只是简单的示例，通常也只是支持本机为单网卡的情况，
未达到应用的级别。
本程序支持本机多网卡(即多IP)，指定IP接收UDP组播数据存储成文件。
而且本程序是多线程并行的工作方式，能保证数据并行接收与存储，从而不会造成数据丢失。
本程序还对UDP组播数据的初始化，接收和结束进行了封装，方便应用开发。
最后说明一下，本程序借鉴ffmpeg的相关代码，完全达到了应用开发级别，
且可以用来理解ffmpeg对UDP数据的处理。

完整的代码开以在这里下载：


一、main.c程序
/******************************************************************************
 * \File
 *   main.c
 * \Brief
 *   Recording udp-multicast data into a file
 * \Author
 *   Hank
 * \Created date
 *   2013-03-12
 ******************************************************************************
 */
#include
#include
#include
#include


#include "udp.h"


#include
#include
#include


/* 
 * 和键盘命令响应相关 
 * 可以参见我的另一篇文章:
 * http://blog.chinaunix.net/uid-26000296-id-3429028.html
 */
#include
#include
#include
#include


static struct termios oldtty;
static int q_pressed = 0;
static int verbose = 1;
static int using_stdin = 0;
static int run_as_daemon = 0;


static volatile int received_sigterm = 0;


static void term_init(void);
static void term_exit(void);
static void sigterm_handler(int sig);
static int read_key(void);




extern char *optarg;
extern int opterr;
struct option opts[] = {
  {"addr", required_argument, NULL, 'i'},
  {"port", required_argument, NULL, 'p'},
  {"host", required_argument, NULL, 's'},
  {"out" , required_argument, NULL, 'o'},
  {"help", required_argument, NULL, 'h'},
  {0,0,0,0}
};


int parse_params(int argc, char** argv, char* host, char* addr, int* port,  char* f);


int main(int argc, char* argv[])
{
  char udp_addr[32]  = "225.1.1.31";
  int  udp_port      = 1234;
  char localhost[32] = "192.168.1.73";
  char filename[512] = "udp.dat";
  
  FILE *fp = NULL;
  int ret;
  int size, sum = 0;


  UDPContext *p_udpctx = NULL;


  char read_buf[4096];
  int key;


  /*Parsing command-line parameters */
  parse_params(argc, argv, localhost, udp_addr, &udp_port, filename);
  fp = fopen(filename, "wb");
  if (fp == NULL)
  {
    printf("Cann't open the file %s\n",filename);
    exit(1);
  }


  /* 处理键盘命令 */
  if (!using_stdin)
  {
    if (verbose >= 0)
      printf("Perss [q] to stop, [?] for help\n");
  }
  term_init();


  /* 初始化UDP组播数据录制 */
  ret = init_udp(&p_udpctx, localhost, udp_addr, udp_port); 


  for(; received_sigterm == 0;)
  {
    float cnt;
    /* if 'q' pressed, exits */
    if (!using_stdin)
    {
      if (q_pressed)
      break;


      /* read_key() returns 0 on EOF */
      key = read_key();
      if (key == 'q')
      {
        printf("quit\n");
        break;
      }
      if (key == '+') 
      {
        verbose++;
        printf("verbose = %d\n", verbose);
      }
      if (key == '-')
      { 
        verbose--;
        printf("verbose = %d\n", verbose);
      }
      if (key == '?')
      {
        printf("key function\n" 
              "? show this help\n" 
              "+ increase verbosity\n" 
              "- decrease verbosity\n" 
              "q quit\n"
              );
      }
    }
    
    /* 将UDP组播数据取出并写到文件 */
    size = read_udp(p_udpctx, read_buf, 4096);
    fwrite(read_buf, size, 1, fp);
    memset(read_buf, 0, sizeof(read_buf));
    sum += size;
    cnt = (float)(sum/1024);


    printf("\rREC: read %d bytes, and write %f kBytes to file.", 
          size, cnt);
    fflush(stdout);
  }


  /* 结束UDP组播数据录制，并回收资源 */
  close_udp(&p_udpctx, localhost);
  fclose(fp);
  
  return 0;
}






int parse_params(int argc, char** argv, 
      char* host, char* addr, int* port,  char* f)
{
  int c, index;


  opterr = 0;
  while ((c = getopt_long(argc, argv, "i:p:s:o:h", opts, NULL)) != -1)
  {
    switch (c)
    {
      case 'i':
        strcpy(addr, optarg);
        break;
      case 'p':
        *port = atoi(optarg);
        break;
      case 's':
        strcpy(host, optarg);
        break;
      case 'o':
        strcpy(f, optarg);
        break;
      case 'h':
      default:
        printf("Usage: \n");
        printf("-i addr : set udp-multicast's ip address\n");
        printf("-p port : set udp-multicast's port\n");
        printf("-s host : set local addresss\n");
        printf("-o file : set output filename\n");
        printf("-h      : print help information\n");
        exit (1);
    }
  }


  /* show banner */
  printf("addr : %s \nport : %d \nhost : %s \nfile : %s\n",
        addr, *port, host, f);


  for (index = optind; index < argc; index++)
  printf("Non-option argument %s\n", argv[index]);


  return 0;
}




static void term_init(void)
{
  if (!run_as_daemon)
  {
    struct termios tty;


    tcgetattr(0, &tty);
    oldtty = tty;
    atexit(term_exit);


    tty.c_iflag &= ~(IGNBRK|BRKINT|PARMRK|ISTRIP
          |INLCR|IGNCR|ICRNL|IXON);
    tty.c_oflag |= OPOST;
    tty.c_lflag &= ~(ECHO|ECHONL|ICANON|IEXTEN);
    tty.c_cflag &= ~(CSIZE|PARENB);
    tty.c_cflag |= CS8;
    tty.c_cc[VMIN] = 1;
    tty.c_cc[VTIME] = 0;


    tcsetattr (0, TCSANOW, &tty);


    /* Quit (POSIX). */
    signal(SIGQUIT, sigterm_handler); 
  }


  signal(SIGINT , sigterm_handler);
  signal(SIGTERM, sigterm_handler); 
}




static void term_exit(void)
{
  printf("%s\n", "TERMINATION");


  if (!run_as_daemon)
  tcsetattr(0, TCSANOW, &oldtty);
  exit(1);
}


static void sigterm_handler(int sig)
{
  received_sigterm = sig;
  q_pressed++;
  term_exit();
}


/* 
 * \Func
 * read_key
 * \Descript
 * read a key without blocking
 */
static int read_key(void)
{
  int n = 1;
  unsigned char ch;
  struct timeval tv;
  fd_set rfds;


  if (run_as_daemon)
  return -1;


  FD_ZERO(&rfds);
  FD_SET(0, &rfds);


  tv.tv_sec = 0;
  tv.tv_usec = 0;
  n = select(1, &rfds, NULL, NULL, &tv);


  if (n > 0)
  {
    n = read(0, &ch, 1);


    if (n == 1)
    return ch;


    return n;
  }


  return -1;
}


二、udp.h头文件
/******************************************************************************
 * \File
 *   udp.h
 * \Brief
 *    
 * \Author
 *   Hank
 * \Created date
 *   2013-03-13
 ******************************************************************************
 */
#ifndef __UDP_H__
#define __UDP_H__
#include
#include


#define MAX_MALLOC_SIZE INT_MAX
#define ALIGN 32


#define UDP_MAX_PKT_SIZE 65536
#define UDP_TX_BUF_SIZE  32768


#define SUCCEED 0
#define FAILED  1


/* 
 * 使用环形FIFO的数据结构存储UDP组播数据 
 * 
 *          ____________________________________
 * buffer->|____________________________________|<-ptr_tail
 *            |                      |           
 *   ptr_read&idx_read        ptr_write/idx_write 
 */
typedef struct UDPFifoBuf {
  uint8_t *buffer, *ptr_tail;     
  uint8_t *ptr_read, *ptr_write;
  uint32_t idx_read, idx_write;
} UDPFifoBuf;


/* UDP上下文数据结构 */
typedef struct UDPContext {
  char addr[32];  // udp-multicast's address
  int  port;      // udp-multicast's port
  char host[32];  // localhost


  int  fd;  
  int  ttl;
  int  pkt_buf_size;
  int  reuse_socket;
  int  is_udp_multicast;


  struct sockaddr_storage udp_addr;
  int    udp_addr_len;


  struct sockaddr_storage my_addr;
  int    my_addr_len;


  UDPFifoBuf   *p_fifo;
  int  cir_buf_size;
  int  cir_buf_err;




  /* Have pthreads*/
  pthread_t thread_cir_buf;   //线程ID


} UDPContext;


int init_udp(UDPContext **pp_udpctx, char *host, char *addr, int port);
int read_udp(UDPContext *p_ctx, char *p_buf, int size);
int close_udp(UDPContext **pp_ctx, char *host);
#endif


三、udp.c程序
/*!
 ******************************************************************************
 * \File
 *   udp.c
 * \Descipt
 *   
 * \Author
 *   Hank
 * \Created date
 *   2013-05-04
 ******************************************************************************
 */
3.1. 头文件与宏定义
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include


#include "udp.h"


extern int errno;
#ifndef IN_MULTICAST
#define IN_MULTICAST(a) ((((uint32_t)(a)) & 0xf0000000) == 0xe0000000)
#endif


#define MAX_CIR_BUF_SIZE (7*188*4096)
#define UDPMIN(a, b) ((a) > (b) ? (b) : (a))


#define AVERROR(e) (-(e))   // Returns a negative error code from a POSIX error code,
                            // to return from library functions.
#define udp_neterrno() AVERROR(errno)