#include <fcntl.h>
#include <stdio.h>
#include <errno.h>
#include <unistd.h>
#include <stdlib.h>
#include <stdarg.h>
#include <strings.h>
#include <sys/socket.h>

extern const char *__progname;

#include "nf.h"
#include "pollloop.h"
#include "scm-rights.h"

#include "link.h"

typedef struct flow FLOW;

struct flow {
  int ffd;
  int tfd;
  const char *fname;
  const char *tname;
  unsigned char buf[8192];
  int bfill;
  int bptr;
  int rid;
  int wid;
  } ;

static int cfd;
static int dfd;
static unsigned char *msg;
static int msglen;
static char *aname;
static char *bname;
static int afd;
static int bfd;
static int nflows;
static char *params;

static void msgspace(int nb)
{
 if (nb > msglen)
  { free(msg);
    msglen = nb;
    msg = malloc(msglen);
  }
}

static void startup_protocol(void)
{
 unsigned char ctlbuf[CMSPACE(2*sizeof(int))];
 struct msghdr mh;
 struct cmsghdr cmh;
 struct iovec iov[2];
 int rv;
 int alen;
 int blen;
 int plen;

 iov[0].iov_base = &alen;
 iov[0].iov_len = sizeof(int);
 iov[1].iov_base = &blen;
 iov[1].iov_len = sizeof(int);
 iov[2].iov_base = &plen;
 iov[2].iov_len = sizeof(int);
 mh.msg_name = 0;
 mh.msg_namelen = 0;
 mh.msg_iov = &iov[0];
 mh.msg_iovlen = 3;
 mh.msg_control = &ctlbuf[0];
 mh.msg_controllen = sizeof(ctlbuf);
 mh.msg_flags = 0;
 rv = recvmsg(cfd,&mh,0);
 if (rv < 0)
  { fprintf(stderr,"%s: link startup recvmsg: %s\n",__progname,strerror(errno));
    exit(0);
  }
 if (rv != 3*sizeof(int))
  { fprintf(stderr,"%s: link startup message size (%d) wrong (not %d)\n",__progname,rv,3*(int)sizeof(int));
    exit(0);
  }
 if (mh.msg_controllen < sizeof(struct cmsghdr))
  { fprintf(stderr,"%s: link startup controllen (%d) too small (< %d)\n",__progname,(int)mh.msg_controllen,(int)sizeof(struct cmsghdr));
    exit(0);
  }
 bcopy(&ctlbuf[0],&cmh,sizeof(cmh));
 if (cmh.cmsg_len < CMLEN(2*sizeof(int)))
  { fprintf(stderr,"%s: link startup cmsg len (%d) too small (< %d)\n",__progname,(int)cmh.cmsg_len,(int)CMLEN(2*sizeof(int)));
    exit(0);
  }
 bcopy(&ctlbuf[CMSKIP(cmh)],&afd,sizeof(int));
 bcopy(&ctlbuf[CMSKIP(cmh)+sizeof(int)],&bfd,sizeof(int));
 msgspace(alen);
 rv = recv(cfd,msg,msglen,0);
 if (rv != alen)
  { fprintf(stderr,"%s: link: startup protocol botch: second recv returned %d, not %d\n",__progname,rv,alen);
    exit(0);
  }
 aname = malloc(alen+1);
 bcopy(msg,aname,alen);
 aname[alen] = '\0';
 msgspace(blen);
 rv = recv(cfd,msg,msglen,0);
 if (rv != blen)
  { fprintf(stderr,"%s: link: startup protocol botch: third recv returned %d, not %d\n",__progname,rv,blen);
    exit(0);
  }
 bname = malloc(blen+1);
 bcopy(msg,bname,blen);
 bname[blen] = '\0';
 msgspace(plen);
 rv = recv(cfd,msg,msglen,0);
 if (rv != plen)
  { fprintf(stderr,"%s: link: startup protocol botch: fourth recv returned %d, not %d\n",__progname,rv,plen);
    exit(0);
  }
 params = malloc(plen+1);
 bcopy(msg,params,plen);
 params[plen] = '\0';
}

static void kill_flow(FLOW *f)
{
 remove_poll_id(f->rid);
 remove_poll_id(f->wid);
 free(f);
 nflows --;
}

static int rtest_flow(void *fv)
{
 FLOW *f;

 f = fv;
 return(f->bptr>=f->bfill);
}

static int wtest_flow(void *fv)
{
 FLOW *f;

 f = fv;
 return(f->bptr<f->bfill);
}

static void rd_flow(void *fv)
{
 FLOW *f;
 int r;

 f = fv;
 r = read(f->ffd,&f->buf[0],sizeof(f->buf));
 if (r < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
     }
    fprintf(stderr,"%s: flow %s -> %s read: %s\n",__progname,f->fname,f->tname,strerror(errno));
    kill_flow(f);
    return;
  }
 if (r == 0)
  { kill_flow(f);
    return;
  }
 f->bfill = r;
 f->bptr = 0;
}

static void wr_flow(void *fv)
{
 FLOW *f;
 int w;

 f = fv;
 w = write(f->ffd,&f->buf[f->bptr],f->bfill-f->bptr);
 if (w < 0)
  { switch (errno)
     { case EINTR:
       case EWOULDBLOCK:
	  return;
     }
    fprintf(stderr,"%s: flow %s -> %s write: %s\n",__progname,f->fname,f->tname,strerror(errno));
    kill_flow(f);
    return;
  }
 if (w == 0)
  { fprintf(stderr,"%s: flow %s -> %s zero write\n",__progname,f->fname,f->tname);
    kill_flow(f);
    return;
  }
 f->bptr += w;
}

static void create_flow(int ffd, int tfd, const char *fname, const char *tname)
{
 FLOW *f;

 f = malloc(sizeof(FLOW));
 f->ffd = ffd;
 f->tfd = tfd;
 f->fname = fname;
 f->tname = tname;
 f->bfill = 0;
 f->bptr = 0;
 f->rid = add_poll_fd(ffd,&rtest_flow,&rwtest_never,&rd_flow,0,f);
 f->wid = add_poll_fd(tfd,&rwtest_never,&wtest_flow,0,&wr_flow,f);
 nflows ++;
}

static int link_die(void *arg __attribute__((__unused__)))
{
 if (nflows < 1)
  { unsigned char msg;
    msg = LINKMSG_DYING;
    nf_send(cfd,&msg,1,0);
    read(dfd,&msg,1);
    exit(0);
  }
 return(BLOCK_NIL);
}

static void set_nonblock(int fd)
{
 fcntl(fd,F_SETFL,fcntl(fd,F_GETFL,0)|O_NONBLOCK);
}

static void rd_link_ctl(void *arg __attribute__((__unused__)))
{
 int n;

 msgspace(1);
 n = recv(cfd,msg,msglen,0);
 if (n < 0)
  { fprintf(stderr,"%s: link %s <-> %s parent control recv: %s\n",__progname,aname,bname,strerror(errno));
    exit(0);
  }
 if (n == 0)
  { fprintf(stderr,"%s: link %s <-> %s empty parent control message\n",__progname,aname,bname);
    exit(0);
  }
 switch (msg[0])
  { case LINKMSG_DIE:
       fprintf(stderr,"%s: link %s <-> %s dying on request\n",__progname,aname,bname);
       exit(0);
       break;
    default:
       fprintf(stderr,"%s: link %s <-> %s bad parent control message %02x\n",__progname,aname,bname,msg[0]);
       exit(0);
       break;
  }
}

static void rd_link_death(void *arg __attribute__((__unused__)))
{
 fprintf(stderr,"%s: link %s <-> %s parent death pipe readable\n",__progname,aname,bname);
 exit(0);
}

static void gen_bufsize(int nb)
{
 unsigned char msg[1+sizeof(int)];

 msg[0] = LINKMSG_BUFSIZE;
 bcopy(&nb,&msg[1],sizeof(int));
 nf_send(cfd,&msg[0],1+sizeof(int),0);
}

static void oprintf(const char *, ...)
	__attribute__((__format__(__printf__,1,2)));
static void oprintf(const char *fmt, ...)
{
 va_list ap;
 char *s;
 int len;
 unsigned char hdr;
 struct iovec iov[2];
 struct msghdr mh;

 va_start(ap,fmt);
 len = vasprintf(&s,fmt,ap);
 va_end(ap);
 gen_bufsize(1+len);
 hdr = LINKMSG_OUTPUT;
 iov[0].iov_base = &hdr;
 iov[0].iov_len = 1;
 iov[1].iov_base = s;
 iov[1].iov_len = len;
 mh.msg_name = 0;
 mh.msg_namelen = 0;
 mh.msg_iov = &iov[0];
 mh.msg_iovlen = 2;
 mh.msg_control = 0;
 mh.msg_controllen = 0;
 mh.msg_flags = 0;
 nf_sendmsg(cfd,&mh,0);
}

static void parse_params(void)
{
 oprintf("params: %s\n",params);
}

void link_main(int cfdarg, int dfdarg)
{
 cfd = cfdarg;
 dfd = dfdarg;
 setproctitle("link");
 msg = 0;
 msglen = 0;
 startup_protocol();
 parse_params();
 msgspace(1+sizeof(int));
 nflows = 0;
 set_nonblock(afd);
 set_nonblock(bfd);
 init_polling();
 add_poll_fd(cfd,&rwtest_always,&rwtest_never,&rd_link_ctl,0,0);
 add_poll_fd(dfd,&rwtest_always,&rwtest_never,&rd_link_death,0,0);
 create_flow(afd,bfd,aname,bname);
 create_flow(bfd,afd,bname,aname);
 add_block_fn(&link_die,0);
 poll_run();
}