Return to repeatts.c CVS log

Up to [DVB] / multiplexer

- fixed rounding bug that led to a non-uniform distribution of the TS
  packets ("XXXXXXX        " instead of "X X X X X X X ")

/*
 * ISO 13818 stream multiplexer / additional repeater tool
 * Copyright (C) 2001 Convergence Integrated Media GmbH Berlin
 * Author: Oskar Schirmer (oskar@convergence.de)
 */

/*
 * Module:  Repeater
 * Purpose: Additional tool to repeat and pipe an input stream.
 *
 * This tool accepts a (time1,time2,filename) tuple via stdin,
 * opens the filename, assuming it contains a ISO 13818 Transport Stream,
 * and sends it packet-wise to stdout timed in such a way, that
 * all pakets are sent equally distributed within time2 msec.
 * As long as no further tuple is provided, after time1 msec the same
 * file is sent over and over again.
 */


#include <stdio.h>
#include "global.h"


#define D(x)                   /* x */

#define MAX_ANOTATE (16 * 256)

static boolean quit;
static int cmdf, outf, nextf;

static int combln, comlln;
static byte combuf[MAX_DATA_COMB];

static int dati, dato;
static byte data[MAX_ANOTATE];

static t_msec nextfdelay;

static int    num_packets;
static t_msec transmission_time;

t_msec
msec_now (void)
{
  static int       first_call = 1;  
  static long long first_time;
  static long long now;

  struct timeval tv;
  
  gettimeofday (&tv, NULL);
#if 0
#define MSEC_EXPONENT 21
  static long last;
  static int local_delta;
  if ((tv.tv_sec & (~((1L << MSEC_EXPONENT) - 1))) != last)
    {
      last = tv.tv_sec & (~((1L << MSEC_EXPONENT) - 1));
      local_delta += 1000 * (1L << MSEC_EXPONENT);
    }
  return ((tv.tv_sec & ((1L << MSEC_EXPONENT) - 1)) * 1000
          + tv.tv_usec / 1000 + local_delta);
#endif
  now = (tv.tv_sec * 1000) + (tv.tv_usec / 1000);
  if (first_call)
    {
      first_call = 0;
      first_time = now;
    }

  return (t_msec)(now - first_time);
}

static void
command_help (char *command)
{
  fprintf (stderr, 
    "Usage:\t%s [<delay> <time> <file>]\n"
    "\tSend <file> repeated every <delay> msec evenly distributed within\n"
    "\t<time> msec, Feed a (delay,time,file) tuple to stdin to reset\n"
    "\tbehaviour, quit program with negative delay, don't repeat with delay=0.\n",
           command);
}

static boolean
line_complete (char **s1, char **s2, char **s3)
{
  int i;
  boolean b = FALSE;
  *s3 = NULL;
  *s2 = NULL;
  *s1 = NULL;
  i = 0;
  while (i < combln)
    {
      if (combuf[i] == '\n')
        {
          comlln = i;
          while (i >= 0)
            {
              if (combuf[i] <= ' ')
                {
                  combuf[i] = 0;
                  b = TRUE;
                }
              else
                {
                  if (b)
                    {
                      *s3 = *s2;
                      *s2 = *s1;
                      b = FALSE;
                    }
                  *s1 = &combuf[i];
                }
              i -= 1;
            }
          return (TRUE);
        }
      i += 1;
    }
  return (FALSE);
}

static boolean
is_long (char *s, long *r)
{
  long i;
  char *e;
  if (s == NULL)
    {
      return (FALSE);
    }
  errno = 0;
  i = strtol (s, &e, 0);
  if ((errno != 0) || (*e != 0))
    {
      return (FALSE);
    }
  *r = i;
  return (TRUE);
}

static boolean
command_do (char *arg1, char *arg2, char *arg3)
{
  long l1, l2;
  struct stat stat;
D (fprintf (stderr, "command_do(%s,%s,%s)\n", arg1, arg2, arg3));
  if (arg1 != NULL)
    {
      if (is_long (arg1, &l1))
        {
          if (l1 < 0)
            {
              quit = TRUE;
              return (TRUE);
            }
          if (arg2 != NULL)
            {
              if (is_long (arg2, &l2))
                {
                  if (l2 >= 0)
                    {
                      if ((l1 >= l2) || (l1 == 0))
                        {
                          if (arg3 != NULL)
                            {
                              if (nextf >= 0)
                                {
                                  close (nextf);
                                }
                              if ((nextf = open (arg3, O_RDONLY)) >= 0)
                                {
                                  if (fstat (nextf, &stat) == 0)
                                    {
D (fprintf (stderr, "file %d, mode %07o, name %s, ino %ld, size %ld\n",
            nextf, stat.st_mode, arg3, stat.st_ino, stat.st_size));
                                      if (S_ISREG (stat.st_mode))
                                        {
                                          if ((stat.st_size %
                                               TS_PACKET_SIZE) == 0)
                                            {
                                              num_packets = stat.st_size /
                                                TS_PACKET_SIZE;
                                              transmission_time = l2;
                                              nextfdelay = l1;
D (fprintf (stderr, "next opened(%d, %d, %d)\n", 
            nextfdelay, num_packets, transmission_time));
                                              return (TRUE);
                                            }
                                          else
                                            {
                                              fprintf (stderr,
                                                "File size not multiple of 188\n");
                                            }
                                        }
                                      else
                                        {
                                          fprintf (stderr,
                                                   "File not regular\n");
                                        }
                                    }
                                  else
                                    {
                                      fprintf (stderr, "Cannot stat file\n");
                                    }
                                  close (nextf);
                                  nextf = -1;
                                }
                              else
                                {
                                  fprintf (stderr, "Cannot open file\n");
                                }
                            }
                          else
                            {
                              fprintf (stderr, "File name missing\n");
                            }
                        }
                      else
                        {
                          fprintf (stderr, "0<delay<time not allowed\n");
                        }
                    }
                  else
                    {
                      fprintf (stderr, "Time must not be negative\n");
                    }
                }
              else
                {
                  fprintf (stderr, "Time must be numeric\n");
                }
            }
          else
            {
              fprintf (stderr, "Time missing\n");
            }
        }
      else
        {
          fprintf (stderr, "Delay must be numeric\n");
        }
    }
  else
    {
      return (TRUE);
    }
  return (FALSE);
}

static boolean
command_init (int cargc, char **cargv)
{
  nextf = -1;
  quit = FALSE;
  combln = 0;
  dati = dato = 0;
  if (cargc > 1)
    {
      if (!strcmp (cargv[1], "--help"))
        {
          command_help (cargv[0]);
          return (FALSE);
        }
      else
        if (!command_do
            (cargv[1], cargc > 2 ? cargv[2] : NULL,
             cargc > 3 ? cargv[3] : NULL))
        {
          command_help (cargv[0]);
          return (FALSE);
        }
    }
  cmdf = STDIN_FILENO;
  outf = STDOUT_FILENO;
  return ((cmdf >= 0) && (outf >= 0));
}

int
main (int argc, char *argv[])
{
  int polli, pollo, polls;
  int toberead = 0;   /* to make gcc happy */
  int currentf;
  int packet_count = 0;

  boolean dotime;
  t_msec rtime;
  t_msec rtime_base = 0;
  t_msec ftime;
  t_msec fdelay = 0;  /* to make gcc happy */
  t_msec now;
  struct pollfd ufds[3];
  if (command_init (argc, &argv[0]))
    {
      currentf = -1;
      rtime = ftime = msec_now ();
      while (!quit)
        {
          now = msec_now ();
          D (fprintf (stderr, "now(%x)\n", now));
          if (currentf < 0)
            {
              toberead = 0;
              fdelay = nextfdelay;
              if ((ftime - now) < 0)
                {
                  ftime = now;
                }
              rtime_base = rtime = ftime;
              packet_count = 0;
              currentf = nextf;
              nextf = -1;
              D (fprintf (stderr, "next current(%d,%d)\n", currentf, fdelay));
            }
          if (currentf >= 0)
            {
              if ((rtime - now) <= 0)
                {
                  if ((((dato - dati - 1) & (MAX_ANOTATE - 1)) - toberead) >
                      TS_PACKET_SIZE)
                    {
                      toberead += TS_PACKET_SIZE;

                      packet_count++;
                      if (num_packets > 0)
                        {
                          rtime = rtime_base + 
                            (t_msec)(((long long)packet_count * transmission_time) /
                                     num_packets);
                          D (fprintf (stderr, "%d %d %d %d %d\n", 
                                      rtime, rtime_base, packet_count, 
                                      transmission_time, num_packets));
                        }
                        
                      dotime = TRUE;
                      D (fprintf (stderr, "timer a(%d,%d)\n", toberead, rtime));
                    }
                  else
                    {
                      rtime_base = rtime = now;
                      packet_count = 0;
                      dotime = FALSE;
                      D (fprintf (stderr, "timer b(%d,%d)\n", toberead, rtime));
                    }
                }
              else
                {
                  dotime = TRUE;
                  D (fprintf (stderr, "timer c(%d,%d)\n", toberead, rtime));
                }
            }
          else
            {
              dotime = FALSE;
              D (fprintf (stderr, "timer c(%d,%d)\n", toberead, rtime));
            }
          ufds[0].fd = cmdf;
          ufds[0].events = POLLIN;
          if (dati != dato)
            {
              ufds[1].fd = outf;
              ufds[1].events = POLLOUT;
              pollo = 1;
            }
          else
            {
              pollo = 0;
            }
          polls = pollo + 1;
          if (toberead > 0)
            {
              polli = polls++;
              ufds[polli].fd = currentf;
              ufds[polli].events = POLLIN;
            }
          else
            {
              polli = 0;
            }
          poll (&ufds[0], polls,
                dotime ? ((rtime - now) > 0) ? (rtime - now) : 0 : -1);
          if (ufds[0].revents & POLLIN)
            {
              char *s1, *s2, *s3;
              if (combln >= MAX_DATA_COMB - HIGHWATER_COM)
                {
                  combln -= HIGHWATER_COM;
                  memmove (&combuf[0], &combuf[HIGHWATER_COM], combln);
                }
              combln += read (cmdf, &combuf[combln], MAX_DATA_COMB - combln);
              while (line_complete (&s1, &s2, &s3))
                {
                  command_do (s1, s2, s3);
                  combln -= comlln;
                  memmove (&combuf[0], &combuf[comlln], combln);
                }
            }
          if ((polli != 0)
              && (ufds[polli].revents & (POLLIN | POLLHUP | POLLERR)))
            {
              int l;
              if (ufds[polli].revents & POLLIN)
                {
                  l = toberead;
                  if (l > (MAX_ANOTATE - dati))
                    {
                      l = MAX_ANOTATE - dati;
                    }
                  l = read (currentf, &data[dati], l);
                  dati = (dati + l) & (MAX_ANOTATE - 1);
                  toberead -= l;
                }
              else
                {
                  l = 0;
                }
              if (l == 0)
                {
                  if ((nextf >= 0) || (fdelay == 0))
                    {
                      close (currentf);
                      currentf = -1;
                    }
                  else
                    {
                      lseek (currentf, 0, SEEK_SET);
                      toberead =
                        ((toberead - 1) / TS_PACKET_SIZE) * TS_PACKET_SIZE;
                    }
                  ftime += fdelay;
                  now = msec_now ();
                  if ((ftime - now) < 0)
                    {
                      ftime = now;
                    }
                  rtime_base = rtime = ftime;
                  packet_count = 0;
                }
            }
          if ((pollo != 0)
              && (ufds[1].revents & (POLLOUT | POLLHUP | POLLERR)))
            {
              if (ufds[1].revents & POLLOUT)
                {
                  int l;
                  if (dati < dato)
                    {
                      l = MAX_ANOTATE - dato;
                    }
                  else
                    {
                      l = dati - dato;
                    }
                  l = write (outf, &data[dato], l);
                  dato = (dato + l) & (MAX_ANOTATE - 1);
                  if (l == 0)
                    {
                      quit = TRUE;
                    }
                }
              else
                {
                  quit = TRUE;
                }
            }
        }
      return (EXIT_SUCCESS);
    }
  return (EXIT_FAILURE);
}