/* $Id: rawmix.c,v 1.1 2023/02/23 06:17:41 urabe Exp $ */
/* "rawmix.c"    2023.2.19 urabe */
/*               modified from raw_raw.c */
/*               2023.2.19-2.23 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/stat.h>

#include <stdio.h>
#include <signal.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <syslog.h>
#include <unistd.h>

#if TIME_WITH_SYS_TIME
#include <sys/time.h>
#include <time.h>
#else  /* !TIME_WITH_SYS_TIME */
#if HAVE_SYS_TIME_H
#include <sys/time.h>
#else  /* !HAVE_SYS_TIME_H */
#include <time.h>
#endif  /* !HAVE_SYS_TIME_H */
#endif  /* !TIME_WITH_SYS_TIME */

#include "daemon_mode.h"
#include "winlib.h"

#define N_HIST    10    /* default length of packet history */
#define DEBUG1     0
#define DEBUG2     0

static const char rcsid[] =
  "$Id: rawmix.c,v 1.1 2023/02/23 06:17:41 urabe Exp $";

static uint8_w ch_table[WIN_CHMAX];
static int n_ch,negate_channel;
static int daemon_mode;

char *progname,*logfile;
int  syslog_mode, exit_status;

/* prototypes */
static void usage(void);
int main(int, char *[]);

struct Shms {
  uint8_w *ptr;      /* read pointer */
  uint8_w *ptr_save; /* saved read pointer */
  uint8_w *ptr_lim;  /* end of data */
  uint32_w size;       /* size of SHM block */
  int tow;        /* 1:with TOW, 0:without TOW, -1:unkown */
  int eobsize;    /* 1:with eobsize, 0:without eobsize, -1:unkown */
  int eobsize_count;
  int c_save;     /* saved couter; -1:initialized */
};

struct ch_hist {
  int n;
  time_t (*ts)[WIN_CHMAX];
  int p[WIN_CHMAX];
};

int
Shm_pre(struct Shm *shr,struct Shms *shrs)
{
  int i;
  uint32_w uni;
  char tb[100];
  if(shrs->ptr==0) { /* initialize */
    if(shr->r<0) return 0; /* no data since initialized */
    /* valid data */
    shrs->ptr=shr->d+shr->r;
    shrs->size=mkuint4(shrs->ptr_save=shrs->ptr);
    if(mkuint4(shrs->ptr+shrs->size-4)==shrs->size) shrs->eobsize=1;
    else shrs->eobsize=0;
    shrs->eobsize_count=shrs->eobsize;
    shrs->tow=-1;
  }
  else if(shrs->ptr==shr->d+shr->p) { /* SHM not advanced */
    return 0;
  }
  else {
    shrs->size=mkuint4(shrs->ptr_save=shrs->ptr);
    if(shrs->size > shr->pl || mkuint4(shrs->ptr+shrs->size-4)==shrs->size) {
      if(++shrs->eobsize_count == 0) shrs->eobsize_count=1;
    } else shrs->eobsize_count=0;
    if((shrs->eobsize && shrs->eobsize_count==0) ||
       (!shrs->eobsize && shrs->eobsize_count>3))
         {write_log("reset(1)");shrs->ptr=0;return 0;}
  }
  if(shrs->eobsize) shrs->ptr_lim=shrs->ptr+shrs->size-4;
  else shrs->ptr_lim=shrs->ptr+shrs->size;
  shrs->c_save=shr->c;
  shrs->ptr+=4; /* skip space for size */
  uni=(uint32_w)(time(NULL)-TIME_OFFSET);
  i=uni-mkuint4(shrs->ptr);
  if(i>=0 && i<1440)   /* with tow */
    {
    if(shrs->tow!=1)
      {
      snprintf(tb,sizeof(tb),"with TOW (diff=%ds)",i);
      write_log(tb);
      if(shrs->tow==0) {write_log("reset(2)");shrs->ptr=0;return 0;}
      shrs->tow=1;
      }
    shrs->ptr+=4;   /* skip space for TOW */
    }
  else if(shrs->tow!=0)
    {
    write_log("without TOW");
    if(shrs->tow==1) {write_log("reset(3)");shrs->ptr=0;return 0;}
    shrs->tow=0;
    }
  return 1;
}
  
int
Shm_post(struct Shm *shr,struct Shms *shrs)
{
  int i;
  if((shrs->ptr=shrs->ptr_save+shrs->size)>shr->d+shr->pl) shrs->ptr=shr->d;
  i=shr->c-shrs->c_save;
  if(!(i<1000000 && i>=0) || mkuint4(shrs->ptr_save)!=shrs->size) {
    write_log("reset");shrs->ptr=0;return 0;
  }
}

static void
usage(void)
{
  WIN_version();
  fprintf(stderr, "%s\n", rcsid);
  fprintf(stderr," usage : '%s (-D) (-d [len(s)]) [in_key1] [in_key2] [out_key] [shm_size(KB)] ([log file]))'\n",progname);
}

int
main(int argc, char *argv[])
  {
  struct Shm  *shmin[2],*shmout;
  struct Shms shmins[2];
  key_t inkey[2],outkey;
  uint8_w *ptw,ts[6]={0,0,0,0,0,0};
  uint32_w uni;
  char tb[256];
  int i,j,k,c,elim,valid[2],ret,in;
  size_t  size_shm, pl_out;
  uint32_w  size;
  WIN_sr  sr;
  WIN_ch ch;
  uint32_w gs;
  int ss;
  static struct ch_hist chhist;
  time_t tts;

  /* extern int optind; */
  /* extern char *optarg; */

  if((progname=strrchr(argv[0],'/')) != NULL) progname++;
  else progname=argv[0];

  daemon_mode = syslog_mode = 0;
  exit_status = EXIT_SUCCESS;
  if(strcmp(progname,"rawmixd")==0) daemon_mode=1;
  chhist.n=N_HIST;

  while((c=getopt(argc,argv,"d:D"))!=-1)
    {
    switch(c)
      {
      case 'd':   /* length of packet history in sec */
        chhist.n=atoi(optarg);
        break;
      case 'D':   /* daemon mode */
	daemon_mode = 1;
        break;
      default:
        fprintf(stderr," option -%c unknown\n",c);
	usage();
        exit(1);
      }
    }
  optind--;
  if(argc<5+optind)
    {
    usage();
    exit(1);
    }

  inkey[0]=atol(argv[1+optind]);
  inkey[1]=atol(argv[2+optind]);
  outkey=atol(argv[3+optind]);
  size_shm=(size_t)atol(argv[4+optind])*1000;
  if(argc>5+optind) logfile=argv[5+optind];
  else
    {
      logfile=NULL;
      if (daemon_mode)
	syslog_mode = 1;
    }

  /* daemon mode */
  if (daemon_mode) {
    daemon_init(progname, LOG_USER, syslog_mode);
    umask(022);
  }

  if(chhist.n>0 && (chhist.ts=(time_t (*)[WIN_CHMAX])
      win_xmalloc(WIN_CHMAX*chhist.n*sizeof(time_t)))==NULL) {
    fprintf(stderr,"malloc failed (chhist.ts)\n");
    exit(1);
  }

  snprintf(tb,sizeof(tb),"n_hist=%d size=%zd",
      chhist.n,WIN_CHMAX*chhist.n*sizeof(time_t));
  write_log(tb);

  /* in shared memory ; SHM key 0 means no use. */
  in=0;
  for(i=0;i<2;i++) {
    if(inkey[i]==0) valid[i]=0;
    else {
      shmin[i] = Shm_read(inkey[i], "in");
      valid[i]=1;
      in++;
    }
  }
  if(in==0) {
    write_log("no input SHM specified");
    end_program();
  }

  /* out shared memory */
  shmout = Shm_create(outkey, size_shm, "out");

  /* initialize buffer */
  Shm_init(shmout, size_shm);
  pl_out = shmout->pl;

  /* sprintf(tb,"start in_key=%d id=%d out_key=%d id=%d size=%d", */
  /*   rawkey,shmid_raw,monkey,shmid_mon,size_shm); */
  /* write_log(tb); */

  signal(SIGTERM,(void *)end_program);
  signal(SIGINT,(void *)end_program);

  shmins[0].ptr=shmins[1].ptr=0;

  ptw=shmout->d+shmout->p;
  for(;;) {
    in=0;
    for(j=0;j<2;j++) {
      if(valid[j]==0) continue;
      ret=Shm_pre(shmin[j],&shmins[j]);
      if(ret) {
#if DEBUG
        for(i=0;i<16;i++) printf("%02X",shmins[j].ptr[i]);
        printf(" : %d in SHM#%d\n",shmins[j].size,j);
#endif
        for(i=5;i>=0;i--) if(ts[i]!=shmins[j].ptr[i]) break;
        if(i>=0) { /* new TS & block */
          /* close last block */
          if((ptw-(shmout->d+shmout->p))>14)
            {
            uni=(uint32_w)(ptw-(shmout->d+shmout->p))+4;
            shmout->d[shmout->p  ]=uni>>24; /* size (H) */
            shmout->d[shmout->p+1]=uni>>16;
            shmout->d[shmout->p+2]=uni>>8;
            shmout->d[shmout->p+3]=uni;     /* size (L) */
            ptw[0]=uni>>24;  /* eobsize (H) */
            ptw[1]=uni>>16;
            ptw[2]=uni>>8;
            ptw[3]=uni;      /* eobsize (L) */
            ptw+=4;
#if DEBUG1
            for(i=0;i<16;i++) printf("%02X",shmout->d[shmout->p+i+8]);
            printf(" : %u in SHMout closed\n",uni);
#endif
            shmout->r=shmout->p;
            if(ptw>shmout->d+shmout->pl) ptw=shmout->d;
            shmout->p=ptw-shmout->d;
            shmout->c++;
            }
          else { /* once opened but no data written */
            ptw=shmout->d+shmout->p;
          }
          /* open new block */
          ptw+=4;          /* size (4) */
          uni=(uint32_w)(time(NULL)-TIME_OFFSET);
          *ptw++=uni>>24;  /* tow (H) */
          *ptw++=uni>>16;
          *ptw++=uni>>8;
          *ptw++=uni;      /* tow (L) */
          for(i=0;i<6;i++) *ptw++=ts[i]=*shmins[j].ptr++;  /* YMDhms (6) */
          tts=check_ts(ts,0,0);
#if DEBUG1
          for(i=0;i<6;i++) printf("%02X",ts[i]);
          printf(" : TS of new block\n");
#endif
        } else shmins[j].ptr+=6;
        do {   /* loop for ch's */
          gs=win_chheader_info(shmins[j].ptr,&ch,&sr,&ss);
#if DEBUG2
          fprintf(stderr,"gs=%d ch=%04X sr=%d ss=%d\n",gs,ch,sr,ss);
#endif
          for(i=0;i<chhist.n;i++) if(chhist.ts[i][ch]==tts) break;
          if(i==chhist.n) {  /* TS not found in last chhist.n packets */
            if(chhist.n>0) {
              chhist.ts[chhist.p[ch]][ch]=tts;
              if(++chhist.p[ch]==chhist.n) chhist.p[ch]=0;
            }
            memcpy(ptw,shmins[j].ptr,gs);
            ptw+=gs;
          }
#if DEBUG1
          else fprintf(stderr,"%5d(!%04X)",gs,ch);
#endif
          shmins[j].ptr+=gs;
        } while(shmins[j].ptr<shmins[j].ptr_lim);
        shmins[j].ptr=shmins[j].ptr_save+shmins[j].size; /* top of next block */
        in++;
      }
      ret=Shm_post(shmin[j],&shmins[j]);
    }
    if(in==0) usleep(10000);
  }
}