/*
 *  dequeue_complex.c
 *
 *    A complex, multithreaded channel program to dequeue queued messages,
 *    producing Batch SMTP (BSMTP) output files.
 *
 *  Copyright (c) 2003, Sun Microsystems, Inc.  All Rights Reserved.
 */

#include <stdio.h>
#include <stdlib.h>
#if !defined(_WIN32)
#include <unistd.h>
#endif
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include "mtasdk.h"

typedef struct {
     int   debug;       /* Debug flag                                         */
     int   max_count;   /* Maximum number of messages per BSMTP file          */
} my_global_context_t;

typedef struct {
     int   id;          /* Dequeue thread's id                                */
     FILE *fp;          /* Dequeue thread's current output file               */
     int   count;       /* Count of messages output by this dequeue thread    */
} my_thread_context_t;


static const char *NotifyToStr(int ret_type, char *buf);
static const char *UniqueName(char *buf, size_t maxbuf, const char *suffix);
static mta_dq_process_done_t process_done;
static mta_dq_process_message_t process_message;


main()
{
     my_global_context_t gctx;
     int ires;

     /*
      *  Initialize the MTA SDK
      */
     if ((ires = mtaInit(0)))
     {
	  mtaLog("mtaInit() returned %d; %s\n", ires, mtaStrError(ires, 0));
	  return(1);
     }

     /*
      *  The global context is shared by all dequeue threads which result
      *  from the same mtaDequeueStart() call.  The global context in this
      *  example provides process_message() with the following
      *
      *    (1) How many messages to put into a BSMTP file before closing
      *        it and starting a new one, and
      *
      *    (2) Whether or not to produce diagnostic debug output.
      */
     gctx.debug     = 1;
     gctx.max_count = 5;

     /*
      *  Start the dequeue loop
      */
     ires = mtaDequeueStart((void *)&gctx, process_message, process_done, 0);

     /*
      *  Check the return status
      */
     if (!ires)
	  /*
	   *  Success
	   */
	  exit(0);

     /*
      *  Produce an error message
      */
     mtaLog("mtaDequeueStart() returned %d; %s", ires, mtaStrError(ires, 0));

     /*
      *  And exit with an error
      */
     exit(1);
}


/*
 *  process_done -- Called by mtaDequeueStart() to clean up and destroy
 *                  a per thread context created by process_message().
 */
static void process_done(void *my_ctx_2, void *my_ctx_1)
{
     my_global_context_t *gctx = (my_global_context_t *)my_ctx_1;
     my_thread_context_t *tctx = (my_thread_context_t *)my_ctx_2;

     if (!tctx)
	  return;

     /*
      *  Generate any requested diagnostic output requested?
      */
     if (gctx && gctx->debug)
	  mtaLog("Dequeue thread done: id=%d; context=%p; messages=%d",
		 tctx->id, tctx, tctx->count);

     /*
      *  Now clean up and destroy the context
      */
     if (tctx->fp)
     {
	  fprintf(tctx->fp, "QUIT\n");
	  fclose(tctx->fp);
     }
     free(tctx);
}


/*
 *  process_message -- Called by mtaDequeueStart() to process a single message.
 */
static int process_message(void **my_ctx_2, void *my_ctx_1, mta_dq_t *dq,
			   const char *env_from, size_t env_from_len)
{
     my_global_context_t *gctx;
     my_thread_context_t *tctx;
     int ires, ret_type;
     const char *to, *env_id, *line;
     size_t len;
     char notify_buf[100];

     /*
      *  This should never happen, but just to be safe we check
      */
     if (!my_ctx_1 || !my_ctx_2)
	  return(MTA_ABORT);

     /*
      *  The pointer to our global context was passed as my_ctx_1
      */
     gctx = (my_global_context_t *)my_ctx_1;

     /*
      *  In this example, we just use the per-thread context to
      *
      *  (1) Track the output file for this dequeue thread across
      *      repeated calls, and
      *
      *  (2) to count how many messages have been output by this
      *      dequeue thread.
      */
     if (!(*my_ctx_2))
     {
	  /*
	   *  First call to process_message() by this dequeue thread.
	   *  Store a pointer to our context.
	   */
	  tctx = (my_thread_context_t *)
	             calloc(1, sizeof(my_thread_context_t));
	  if (!tctx)
	       /*
		*  Insufficient virtual memory -- might as well
		*  defer processing of this message AND delete this thread
		*/
	       return(MTA_ABORT);
	  *my_ctx_2 = (void *)tctx;

	  /*
	   *  Debug output?
	   */
	  if (gctx->debug)
	  {
	       tctx->id = mtaDequeueThreadId(dq);
	       mtaLog("Dequeue thread starting: id=%d; context=%p",
		      tctx->id, tctx);
	  }
     }
     else
	  /*
	   *  This dequeue thread has already called process_message()
	   *  previously.
	   */
	  tctx = (my_thread_context_t *)(*my_ctx_2);

     /*
      *  Send a HELO or a RSET?
      */
     if (0 == (tctx->count % gctx->max_count))
     {
	  char buf[1024];
	  int fd;

	  /*
	   *  Need to send a HELO
	   */

	  /*
	   *  Send a QUIT if we've already sent a HELO previously
	   */
	  if (tctx->count > 0 && tctx->fp)
	  {
	       fprintf(tctx->fp, "QUIT\n");
	       fclose(tctx->fp);
	       tctx->fp = NULL;
	  }

	  /*
	   *  Now open a file
	   */
	  fd = open(UniqueName(buf, sizeof(buf), ".bsmtp"),
		    O_WRONLY | O_CREAT | O_EXCL, 0770);
	  if (fd < 0 || !(tctx->fp = fdopen(fd, "w")))
	  {
	       /*
		*  Unable to open the output file
		*  Defer processing of this message
		*/
	       if (gctx->debug)
		    mtaLog("Unable to open the output file %s; errno=%d; %s",
			   buf, errno, strerror(errno));
	       if (fd >= 0)
	       {
		    /*
		     *  The open() succeeded, but the fdopen() failed
		     */
		    close(fd);
		    remove(buf);
	       }
	       return(MTA_NO);
	  }

	  /*
	   *  Now send the HELO
	   */
	  fprintf(tctx->fp, "HELO %s\n",
		  mtaChannelToHost(NULL, NULL, MTA_DQ_CONTEXT, dq, 0));
     }
     else
     {
	  /*
	   *  We've already sent a HELO
	   *  Send a RSET to start a new message
	   */
	  fprintf(tctx->fp, "RSET\n");
     }
     tctx->count++;

     /*
      *  Output the command
      *     MAIL FROM: <from-adr> RET=return-type ENVID=id
      */
     env_id   = NULL;
     ret_type = MTA_NOTIFY_DEFAULT;
     mtaDequeueInfo(dq, MTA_ENV_ID, &env_id, NULL,
		    MTA_NOTIFY_FLAGS, &ret_type, 0);
     fprintf(tctx->fp, "MAIL FROM:<%s> RET=%s%s%s\n", env_from,
	     NotifyToStr(ret_type, NULL),
	     (env_id ? " ENVID=" : ""), (env_id ? env_id : ""));

     /*
      *  Output the command
      *     RCPT TO: <to-adr> NOTIFY=notify-type
      *  for each recipient address
      */
     while (!(ires = mtaDequeueRecipientNext(
	                              dq, &to, &len,
				      MTA_NOTIFY_FLAGS, &ret_type, 0)))
     {
	  fprintf(tctx->fp, "RCPT TO:<%s> NOTIFY=%s\n",
		  to, NotifyToStr(ret_type, notify_buf));
	  /*
	   *  Indicate that delivery to this recipient succeeded
	   */
	  mtaDequeueRecipientDisposition(dq, to, len,
					 MTA_DISP_DELIVERED, 0);
     }

     /*
      *  If ires == MTA_EOF, then we exited the loop normally;
      *  otherwise, there's been an error of some sort.
      */
     if (ires != MTA_EOF)
	  /*
	   *  An error -- defer processing of this message
	   */
	  return(ires);

     /*
      *  Now output the message itself
      */
     fprintf(tctx->fp, "DATA\n");
     while (!(ires = mtaDequeueLineNext(dq, &line, &len)))
     {
	  /*
	   *  Check to see if we need to dot-stuff the line
	   */
	  if (len == 1 && line[0] == '.')
	       fprintf(tctx->fp, ".");

	  /*
	   *  Now output the line
	   *  Note that the line buffer may not be NULL terminated!
	   */
	  fprintf(tctx->fp, "%.*s\n", len, line);
     }

     /*
      *  If ires == MTA_EOF, then we exited the loop normally;
      *  otherwise, there's been an error of some sort.
      */
     if (ires != MTA_EOF)
	  /*
	   *  An error occurred -- defer processing of this message
	   */
	  return(ires);

     /*
      *  Output the "." command to terminate this message
      */
     fprintf(tctx->fp, ".\n");

     /*
      *  And dequeue the message
      */
     ires = mtaDequeueMessageFinish(dq, 0);

     /*
      *  All done; might as well return ires as our result
      */
     return(ires);
}


/*
 *  NotifyToStr -- Convert a bitmask of MTA_NOTIFY_ flags to a readable string.
 */
static const char *
NotifyToStr(int ret_type, char *buf)
{
     if (!buf)
	  /*
	   *  Doing a RET= parameter to a MAIL FROM command
	   */
	  return((ret_type & MTA_NOTIFY_CONTENT_FULL) ?
		                             "FULL" : "HDRS");

     buf[0] = '\0';

     if (ret_type & MTA_NOTIFY_SUCCESS)
	  strcat(buf, "SUCCESS");

     if (ret_type & MTA_NOTIFY_FAILURE)
     {
	  if (buf[0])
	       strcat(buf, ",");
	  strcat(buf, "FAILURE");
     }

     if (ret_type & MTA_NOTIFY_DELAY)
     {
	  if (buf[0])
	       strcat(buf, ",");
	  strcat(buf, "DELAY");
     }

     if (!buf[0])
	  strcat(buf, "NEVER");

     return(buf);
}


/*
 *  UniqueName -- Generate a unique string suitable for use as a file name.
 */
static const char *
UniqueName(char *buf, size_t maxbuf, const char *suffix)
{
     strcpy(buf, "/tmp/");
     mtaUniqueString(buf+5, NULL, maxbuf-5);
     strcat(buf, suffix);
     return(buf);
}
