subscribe.c

Go to the documentation of this file.

#include <unistd.h>
#include "str.h"
#include "stralloc.h"
#include "getln.h"
#include "readwrite.h"
#include "buffer.h"
#include "logmsg.h"
#include "open.h"
#include "byte.h"
#include "case.h"
#include "lock.h"
#include "error.h"
#include "subscribe.h"
#include "uint_t.h"
#include "fmt.h"
#include "errtxt.h"
#include "logaddr.h"
#include "idx.h"
#include "lockfile.h"
 
#define WHO "subscribe"

 
static stralloc addr = {0};
static stralloc lcaddr = {0};
static stralloc line = {0};
static stralloc fnnew = {0};
static stralloc fn = {0};
static stralloc fnlock = {0};
 
static void die_nomem() { logmsg(WHO,111,FATAL,ERR_NOMEM); }
static void die_read() { logmsg(WHO,111,FATAL,B(ERR_READ,fn.s)); }
static void die_write() { logmsg(WHO,111,FATAL,B(ERR_WRITE,fnnew.s)); }
 
static int fd;
static buffer bi;
static char inbuf[256];
static int fdnew;
static buffer bo;
static char outbuf[256];

 
int subscribe(const char *dbname,const char *userhost,int flagadd,const char *comment,
              const char *event,int flagmysql,int forcehash,const char *tab)
{
  int fdlock;
 
  unsigned int j;
  uint32 h,lch;
  unsigned char ch,lcch;
  int match;
  int flagwasthere;
 
  if (userhost[str_chr(userhost,'\n')])
    logmsg(WHO,100,FATAL,ERR_ADDR_NL);
 
    if (!stralloc_copys(&addr,"T")) die_nomem();
    if (!stralloc_cats(&addr,userhost)) die_nomem();
    if (addr.len > 401)
      logmsg(WHO,100,FATAL,ERR_ADDR_LONG);
 
    j = byte_rchr(addr.s,addr.len,'@');
    if (j == addr.len)
      logmsg(WHO,100,FATAL,ERR_ADDR_AT);
 
    case_lowerb(addr.s + j + 1,addr.len - j - 1);
    if (!stralloc_copy(&lcaddr,&addr)) die_nomem();
    case_lowerb(lcaddr.s + 1,j - 1);             /* make all-lc version of address */
 
    if (forcehash >= 0 && forcehash <= 52) {
      ch = lcch = (unsigned char) forcehash;
    } else {
      h = 5381;
      lch = h;
      for (j = 0;j < addr.len;++j) {
        h = (h + (h << 5)) ^ (uint32) (unsigned char) addr.s[j];
        lch = (lch + (lch << 5)) ^ (uint32) (unsigned char) lcaddr.s[j];
      }
      lcch = 64 + (lch % 53);
      ch = 64 + (h % 53);
    }
 
    if (!stralloc_0(&addr)) die_nomem();
    if (!stralloc_0(&lcaddr)) die_nomem();
    if (!stralloc_copys(&fn,dbname)) die_nomem();
    if (!stralloc_copys(&fnlock,dbname)) die_nomem();
 
    if (!stralloc_cats(&fn,"/subscribers/")) die_nomem();
    if (!stralloc_catb(&fn,&lcch,1)) die_nomem();
    if (!stralloc_copy(&fnnew,&fn)) die_nomem();
    /* code later depends on fnnew = fn + 'n' */
    if (!stralloc_cats(&fnnew,"n")) die_nomem();
    if (!stralloc_cats(&fnlock,"/lock")) die_nomem();
    if (!stralloc_0(&fnnew)) die_nomem();
    if (!stralloc_0(&fn)) die_nomem();
    if (!stralloc_0(&fnlock)) die_nomem();
 
    fdlock = lockfile(fnlock.s);
 
    /* do lower case hashed version first */
 
    fdnew = open_trunc(fnnew.s);
    if (fdnew == -1) die_write();
    buffer_init(&bo,buffer_unixwrite,fdnew,outbuf,sizeof(outbuf));
 
    flagwasthere = 0;
 
    fd = open_read(fn.s);
    if (fd == -1) {
      if (errno != ENOENT) { close(fdnew); die_read(); }
    }
    else {
      buffer_init(&bi,buffer_unixread,fd,inbuf,sizeof(inbuf));
 
      for (;;) {
        if (getln(&bi,&line,&match,'\0') == -1) {
          close(fd); close(fdnew); die_read();
        }
        if (!match) break;
        if (line.len == addr.len)
          if (!case_diffb(line.s,line.len,addr.s)) {
            flagwasthere = 1;
            if (!flagadd) continue;
        }
        if (buffer_put(&bo,line.s,line.len) == -1) {
          close(fd); close(fdnew); die_write();
        }
      }
 
      close(fd);
    }
 
    if (flagadd && !flagwasthere)
      if (buffer_put(&bo,addr.s,addr.len) == -1) {
        close(fdnew); die_write();
      }
 
    if (buffer_flush(&bo) == -1) { close(fdnew); die_write(); }
    if (fsync(fdnew) == -1) { close(fdnew); die_write(); }
    close(fdnew);
 
    if (rename(fnnew.s,fn.s) == -1)
      logmsg(WHO,111,FATAL,B(ERR_MOVE,fnnew.s," to : ",fn.s));
 
    if ((ch == lcch) || flagwasthere) {
      close(fdlock);
      if (flagadd ^ flagwasthere) {
        if (!stralloc_0(&addr)) die_nomem();
        logaddr(dbname,event,addr.s+1,comment);
        return 1;
      }
      return 0;
    }
 
    /* If unsub and not found and hashed differ, OR */
    /* sub and not found (so added with new hash) */
    /* do the 'case-dependent' hash */
 
    fn.s[fn.len - 2] = ch;
    fnnew.s[fnnew.len - 3] = ch;
    fdnew = open_trunc(fnnew.s);
    if (fdnew == -1) die_write();
    buffer_init(&bo,buffer_unixwrite,fdnew,outbuf,sizeof(outbuf));
 
    fd = open_read(fn.s);
    if (fd == -1) {
      if (errno != ENOENT) { close(fdnew); die_read(); }
    } else {
      buffer_init(&bi,buffer_unixread,fd,inbuf,sizeof(inbuf));
 
      for (;;) {
        if (getln(&bi,&line,&match,'\0') == -1)
          { close(fd); close(fdnew); die_read(); }
        if (!match) break;
        if (line.len == addr.len)
          if (!case_diffb(line.s,line.len,addr.s)) {
            flagwasthere = 1;
            continue;                  /* always want to remove from case-sensitive hash */
          }
        if (buffer_put(&bo,line.s,line.len) == -1)
          { close(fd); close(fdnew); die_write(); }
      }
 
      close(fd);
    }
 
    if (buffer_flush(&bo) == -1) { close(fdnew); die_write(); }
    if (fsync(fdnew) == -1) { close(fdnew); die_write(); }
    close(fdnew);
 
    if (rename(fnnew.s,fn.s) == -1)
      logmsg(WHO,111,FATAL,B(ERR_MOVE,fnnew.s," to :",fn.s));
 
    close(fdlock);
    if (flagadd ^ flagwasthere) {
      if (!stralloc_0(&addr)) die_nomem();
      logaddr(dbname,event,addr.s+1,comment);
      return 1;
    }
    return 0;
 
}