mbx.c

mgcp协议源代码。支持多种编码：g711

	mbx_abort (stream);
	return NIL;
      }
				/* note file offset of header */
      elt->private.special.offset = curpos;
				/* and internal header size */
      elt->private.special.text.size = i;
				/* header size not known yet */
      elt->private.msg.header.text.size = 0;
      elt->rfc822_size = j;	/* note message size */
				/* calculate system flags */
      if (k & fSEEN) elt->seen = T;
      if (k & fDELETED) elt->deleted = T;
      if (k & fFLAGGED) elt->flagged = T;
      if (k & fANSWERED) elt->answered = T;
      if (k & fDRAFT) elt->draft = T;
      t[8] = '\0';		/* get user flags value */
      elt->user_flags = strtoul (t,NIL,16);
				/* UID already assigned? */
      if (!(elt->private.uid = m)) {
	elt->recent = T;	/* no, mark as recent */
	++recent;		/* count up a new recent message */
				/* assign new UID */
	elt->private.uid = ++stream->uid_last;
	mbx_update_status (stream,elt->msgno,NIL);
      }
				/* update last parsed UID */
      lastuid = elt->private.uid;
    }
    curpos += i + j;		/* update position */
  }

				/* update header */
  if (!stream->rdonly) mbx_update_header (stream);
  fsync (LOCAL->fd);		/* make sure all the UID updates take */
				/* update parsed file size and time */
  LOCAL->filesize = sbuf.st_size;
  fstat (LOCAL->fd,&sbuf);	/* get status again to ensure time is right */
  LOCAL->filetime = sbuf.st_mtime;
  stream->silent = silent;	/* can pass up events now */
  mail_exists (stream,nmsgs);	/* notify upper level of new mailbox size */
  mail_recent (stream,recent);	/* and of change in recent messages */
  return LONGT;			/* return the winnage */
}

/* MBX get cache element with status updating from file
 * Accepts: MAIL stream
 *	    message number
 *	    expunge OK flag
 * Returns: cache element
 */

MESSAGECACHE *mbx_elt (MAILSTREAM *stream,unsigned long msgno,long expok)
{
  MESSAGECACHE *elt = mail_elt (stream,msgno);
  struct {			/* old flags */
    unsigned int seen : 1;
    unsigned int deleted : 1;
    unsigned int flagged : 1;
    unsigned int answered : 1;
    unsigned int draft : 1;
    unsigned long user_flags;
  } old;
  old.seen = elt->seen; old.deleted = elt->deleted; old.flagged = elt->flagged;
  old.answered = elt->answered; old.draft = elt->draft;
  old.user_flags = elt->user_flags;
				/* get new flags */
  if (mbx_read_flags (stream,elt) && expok) {
    mail_expunged (stream,elt->msgno);
    return NIL;			/* return this message was expunged */
  }
  if ((old.seen != elt->seen) || (old.deleted != elt->deleted) ||
      (old.flagged != elt->flagged) || (old.answered != elt->answered) ||
      (old.draft != elt->draft) || (old.user_flags != elt->user_flags))
    mm_flags (stream,msgno);	/* let top level know */
  return elt;
}

/* MBX read flags from file
 * Accepts: MAIL stream
 *	    cache element
 * Returns: non-NIL if message expunged
 */

unsigned long mbx_read_flags (MAILSTREAM *stream,MESSAGECACHE *elt)
{
  unsigned long i;
				/* noop if readonly and have valid flags */
  if (stream->rdonly && elt->valid) return NIL;
				/* set the seek pointer */
  lseek (LOCAL->fd,(off_t) elt->private.special.offset +
	 elt->private.special.text.size - 23,L_SET);
				/* read the new flags */
  if (read (LOCAL->fd,LOCAL->buf,12) < 0) {
    sprintf (LOCAL->buf,"Unable to read new status: %s",strerror (errno));
    fatal (LOCAL->buf);
  }
  LOCAL->buf[12] = '\0';	/* tie off buffer */
				/* calculate system flags */
  i = strtoul (LOCAL->buf+8,NIL,16);
  elt->seen = i & fSEEN ? T : NIL;
  elt->deleted = i & fDELETED ? T : NIL;
  elt->flagged = i & fFLAGGED ? T : NIL;
  elt->answered = i & fANSWERED ? T : NIL;
  elt->draft = i & fDRAFT ? T : NIL;
  LOCAL->expunged |= i & fEXPUNGED ? T : NIL;
  LOCAL->buf[8] = '\0';		/* tie off flags */
				/* get user flags value */
  elt->user_flags = strtoul (LOCAL->buf,NIL,16);
  elt->valid = T;		/* have valid flags now */
  return i & fEXPUNGED;
}

/* MBX update header
 * Accepts: MAIL stream
 */

void mbx_update_header (MAILSTREAM *stream)
{
  int i;
  char *s = LOCAL->buf;
  memset (s,'\0',HDRSIZE);	/* initialize header */
  sprintf (s,"*mbx*\015\012%08lx%08lx\015\012",
	   stream->uid_validity,stream->uid_last);
  for (i = 0; (i < NUSERFLAGS) && stream->user_flags[i]; ++i)
    sprintf (s += strlen (s),"%s\015\012",stream->user_flags[i]);
  LOCAL->ffuserflag = i;	/* first free user flag */
				/* can we create more user flags? */
  stream->kwd_create = (i < NUSERFLAGS) ? T : NIL;
				/* write reserved lines */
  while (i++ < NUSERFLAGS) strcat (s,"\015\012");
  while (T) {
    lseek (LOCAL->fd,0,L_SET);	/* rewind file */
				/* write new header */
    if (write (LOCAL->fd,LOCAL->buf,HDRSIZE) > 0) break;
    mm_notify (stream,strerror (errno),WARN);
    mm_diskerror (stream,errno,T);
  }
}

/* MBX update status string
 * Accepts: MAIL stream
 *	    message number
 *	    flags
 */

void mbx_update_status (MAILSTREAM *stream,unsigned long msgno,long flags)
{
  MESSAGECACHE *elt = mail_elt (stream,msgno);
  struct stat sbuf;
  int expflag;
				/* readonly */
  if (stream->rdonly || !elt->valid) mbx_read_flags (stream,elt);
  else {			/* readwrite */
				/* want to expunge message? */
    if (elt->deleted && (flags & mus_EXPUNGE)) expflag = fEXPUNGED;
    else {			/* seek to system flags */
      lseek (LOCAL->fd,(off_t) elt->private.special.offset +
	     elt->private.special.text.size - 15,L_SET);
				/* read the current system flags */
      if (read (LOCAL->fd,LOCAL->buf,4) < 0) {
	sprintf (LOCAL->buf,"Unable to read system flags: %s",
		 strerror (errno));
	fatal (LOCAL->buf);
      }
      LOCAL->buf[4] = '\0';	/* tie off buffer */
				/* note current set of expunged flag */
      expflag = (strtoul (LOCAL->buf,NIL,16)) & fEXPUNGED;
    }
				/* print new flag string */
    sprintf (LOCAL->buf,"%08lx%04x-%08lx",elt->user_flags,(unsigned)
	     (expflag + (fSEEN * elt->seen) + (fDELETED * elt->deleted) +
	      (fFLAGGED * elt->flagged) + (fANSWERED * elt->answered) +
	      (fDRAFT * elt->draft)),elt->private.uid);
    while (T) {			/* get to that place in the file */
      lseek (LOCAL->fd,(off_t) elt->private.special.offset +
	     elt->private.special.text.size - 23,L_SET);
				/* write new flags and UID */
      if (write (LOCAL->fd,LOCAL->buf,21) > 0) break;
      mm_notify (stream,strerror (errno),WARN);
      mm_diskerror (stream,errno,T);
    }
    if (flags & mus_SYNC) {	/* sync if requested */
      fsync (LOCAL->fd);
      fstat (LOCAL->fd,&sbuf);	/* get new write time */
      LOCAL->filetime = sbuf.st_mtime;
    }
  }
}

/* MBX locate header for a message
 * Accepts: MAIL stream
 *	    message number
 *	    pointer to returned header size
 *	    pointer to possible returned header
 * Returns: position of header in file
 */

#define HDRBUFLEN 4096		/* good enough for most headers */
#define SLOP 4			/* CR LF CR LF */

unsigned long mbx_hdrpos (MAILSTREAM *stream,unsigned long msgno,
			  unsigned long *size,char **hdr)
{
  unsigned long siz;
  long i;
  char *s,*t,*te;
  MESSAGECACHE *elt = mbx_elt (stream,msgno,NIL);
  unsigned long ret = elt->private.special.offset +
    elt->private.special.text.size;
  if (hdr) *hdr = NIL;		/* assume no header returned */
				/* is header size known? */ 
  if (*size = elt->private.msg.header.text.size) return ret;
				/* paranoia check */
  if (LOCAL->buflen < (HDRBUFLEN + SLOP)) {
    fs_give ((void **) &LOCAL->buf);
    LOCAL->buf = (char *) fs_get ((LOCAL->buflen = HDRBUFLEN) + SLOP);
  }
  lseek (LOCAL->fd,ret,L_SET);	/* get to header position */
  for (siz = 0, s = LOCAL->buf;	/* read HDRBUFLEN chunks with 4 byte slop */
       (i = min ((long) (elt->rfc822_size - siz),(long) HDRBUFLEN)) &&
       (read (LOCAL->fd,s,i) == i);
       siz += (t - LOCAL->buf) - SLOP, s = LOCAL->buf + SLOP) {
    te = (t = s + i) - 12;	/* calculate end of fast scan */
				/* fast scan for CR */
    for (s = LOCAL->buf; s < te;)
      if (((*s++ == '\015') || (*s++ == '\015') || (*s++ == '\015') ||
	   (*s++ == '\015') || (*s++ == '\015') || (*s++ == '\015') ||
	   (*s++ == '\015') || (*s++ == '\015') || (*s++ == '\015') ||
	   (*s++ == '\015') || (*s++ == '\015') || (*s++ == '\015')) &&
	  (*s == '\012') && (*++s == '\015') && (*++s == '\012')) {
	*size = elt->private.msg.header.text.size = siz + (++s - LOCAL->buf);
	if (hdr) *hdr = LOCAL->buf;
	return ret;
      }
    for (te = t - 3; (s < te);)	/* final character-at-a-time scan */
      if ((*s++ == '\015') && (*s == '\012') && (*++s == '\015') &&
	  (*++s == '\012')) {
	*size = elt->private.msg.header.text.size = siz + (++s - LOCAL->buf);
	if (hdr) *hdr = LOCAL->buf;
	return ret;
      }
    if (i <= SLOP) break;	/* end of data */
				/* slide over last 4 bytes */
    memmove (LOCAL->buf,t - SLOP,SLOP);
    hdr = NIL;			/* can't return header this way */
  }
				/* not found: header consumes entire message */
  elt->private.msg.header.text.size = *size = elt->rfc822_size;
  if (hdr) *hdr = LOCAL->buf;	/* possibly return header too */
  return ret;
}

/* MBX mail rewrite mailbox
 * Accepts: MAIL stream
 *	    pointer to return reclaimed size
 *	    flags (non-NIL to do expunge)
 * Returns: number of expunged messages
 */

unsigned long mbx_rewrite (MAILSTREAM *stream,unsigned long *reclaimed,
			   long flags)
{
  struct stat sbuf;
  off_t pos,ppos;
  int ld;
  unsigned long i,j,k,m,n,delta;
  unsigned long recent = 0;
  char lock[MAILTMPLEN];
  MESSAGECACHE *elt;
  blocknotify_t bn = (blocknotify_t) mail_parameters (NIL,GET_BLOCKNOTIFY,NIL);
  /* The cretins who designed flock() created a window of vulnerability in
   * upgrading locks from shared to exclusive or downgrading from exclusive
   * to shared.  Rather than maintain the lock at shared status at a minimum,
   * flock() actually *releases* the former lock.  Obviously they never talked
   * to any database guys.  Fortunately, we have the parse/append permission
   * lock.  If we require this lock before going exclusive on the mailbox,
   * another process can not sneak in and steal the exclusive mailbox lock on
   * us, because it will block on trying to get parse/append permission first.
   */
				/* get parse/append permission */
  if ((ld = lockfd (LOCAL->fd,lock,LOCK_EX)) < 0) {
    mm_log ("Unable to lock mailbox for rewrite",ERROR);
    return *reclaimed = 0;
  }

				/* get exclusive access */
  if (!flock (LOCAL->fd,LOCK_EX|LOCK_NB)) {
    mm_critical (stream);	/* go critical */
    for (i = 1,n = delta = *reclaimed = 0,pos = ppos = HDRSIZE;
	 i <= stream->nmsgs; ) {
      elt = mbx_elt (stream,i,NIL);
				/* note if message not at predicted location */
      if (m = elt->private.special.offset - ppos) {
	ppos = elt->private.special.offset;
	*reclaimed += m;	/* note reclaimed message space */
	delta += m;		/* and as expunge delta  */
      }
				/* number of bytes to smash or preserve */
      ppos += (k = elt->private.special.text.size + elt->rfc822_size);
				/* if deleted */
      if (flags && elt->deleted) {
	delta += k;		/* number of bytes to delete */
	mail_expunged(stream,i);/* notify upper levels */
	n++;			/* count up one more expunged message */
      }
      else if (i++ && delta) {	/* preserved message */
	if (elt->recent) ++recent;
				/* first byte to preserve */
	j = elt->private.special.offset;
	do {			/* read from source position */
	  m = min (k,LOCAL->buflen);
	  lseek (LOCAL->fd,j,L_SET);
	  read (LOCAL->fd,LOCAL->buf,m);
	  pos = j - delta;	/* write to destination position */
	  while (T) {
	    lseek (LOCAL->fd,pos,L_SET);
	    if (write (LOCAL->fd,LOCAL->buf,m) > 0) break;
	    mm_notify (stream,strerror (errno),WARN);
	    mm_diskerror (stream,errno,T);
	  }
	  pos += m;		/* new position */
	  j += m;		/* next chunk, perhaps */
	} while (k -= m);	/* until done */
				/* note the new address of this text */
	elt->private.special.offset -= delta;
      }
				/* preserved but no deleted messages yet */
      else pos = elt->private.special.offset + k;
    }
				/* deltaed file size match position? */
    if (m = (LOCAL->filesize -= delta) - pos) {
      *reclaimed += m;		/* probably an fEXPUNGED msg */
      LOCAL->filesize = pos;	/* set correct size */
    }
				/* truncate file after last message */
    ftruncate (LOCAL->fd,LOCAL->filesize);
    fsync (LOCAL->fd);		/* force disk update */
    mm_nocritical (stream);	/* release critical */
    (*bn) (BLOCK_FILELOCK,NIL);
    flock (LOCAL->fd,LOCK_SH);	/* allow sharers again */
    (*bn) (BLOCK_NONE,NIL);
    unlockfd (ld,lock);		/* release exclusive parse/append permission */
  }

  else {			/* can't get exclusive */
    (*bn) (BLOCK_FILELOCK,NIL);
    flock (LOCAL->fd,LOCK_SH);	/* recover previous shared mailbox lock */
    (*bn) (BLOCK_NONE,NIL);
    unlockfd (ld,lock);		/* release exclusive parse/append permission */
				/* checkpoint while shared? */
    if (!flags) n = *reclaimed = 0;
				/* no, do hide-expunge */
    else for (i = 1,n = *reclaimed = 0; i <= stream->nmsgs; ) {
      if (elt = mbx_elt (stream,i,T)) {
	if (elt->deleted) {	/* make deleted message invisible */
	  mbx_update_status (stream,elt->msgno,mus_EXPUNGE);
				/* notify upper levels */
	  mail_expunged (stream,i);
	  n++;			/* count up one more expunged message */
	}
	else {
	  i++;			/* preserved message */
	  if (elt->recent) ++recent;
	}
      }
      else n++;			/* count up one more expunged message */
    }
    fsync (LOCAL->fd);		/* force disk update */
  }
  fstat (LOCAL->fd,&sbuf);	/* get new write time */
  LOCAL->filetime = sbuf.st_mtime;
				/* notify upper level of new mailbox size */
  mail_exists (stream,stream->nmsgs);
  mail_recent (stream,recent);
  return n;			/* return number of expunged messages */
}