Fixes in stream subsystem. Rewrite and optimize maidag lmtp mode using streams.

* libmailutils/file_stream.c (mu_fd_stream_create): Mark stream as
open, do not call mu_stream_open explicitly.
* libmailutils/message_stream.c (mu_stream_to_message): Bugfixes,
wrong owner given to mu_envelope_set_ calls.
* libmailutils/stream.c (_MU_STR_FLUSH_ALL)
(_MU_STR_FLUSH_KEEP): New macros for _stream_flush_buffer.
(_stream_flush_buffer): Change the meaning of the last argument.
All callers updated.
(mu_stream_seek): Fix operation with MU_SEEK_END.
Call _mu_stream_cleareof on success.
(mu_stream_read): Call _stream_flush_buffer in buffered mode.
(mu_stream_getdelim, mu_stream_readdelim): Call _stream_flush_buffer.
* libmailutils/streamcpy.c (mu_stream_copy): Reset size if mu_stream_seek
fails.
* libmailutils/temp_file_stream.c (mu_temp_file_stream_create): Set
full buffering mode by default.

* maidag/mailtmp.c: Remove.
* maidag/Makefile.am (maidag_SOURCES): Remove mailtmp.c
* po/POTFILES.in: Likewise.
* maidag/deliver.c (make_tmp): Rewrite.  Return mu_mailbox_t.
All callers changed.
* maidag/lmtp.c (lmtp_transcript): Remove static.
(lmpt_transcript): New function.
(lmtp_reply): Use mu_stream_t instead of FILE.
(xlatnl): Remove.  Superseded by mu_rtrim_cset and family.
(mtmp, mbox): Remove globals.
(mesg): New global.
(cfun_unknown, cfun_mail_from, cfun_rcpt_to)
(dot_temp_fail, dot_deliver, cfun_rset)
(cfun_lhlo, cfun_quit, cfun_help): Use mu_stream_t instead of FILE.
(cfun_data): Rewrite.
(cfun_dot): Remove.
(to_fgets): Rewrite using mu_stream_t.
(lmtp_loop): Change signature. Rewrite using mu_stream_t.
(lmtp_connection, maidag_lmtp_server): Update accordingly.
* maidag/maidag.c (maidag_transcript): New global.
(options, parse_opt): New option --transcript.
* maidag/maidag.h (maidag_transcript): New extern.
(mail_tmp_begin, mail_tmp_add_line, mail_tmp_finish)
(mail_tmp_destroy): Remove.

12 files changed, 407 insertions, 423 deletions

diff --git a/libmailutils/file_stream.c b/libmailutils/file_stream.c
index 578f4829e..e6a6e67fd 100644
--- a/libmailutils/file_stream.c
+++ b/libmailutils/file_stream.c
@@ -387,20 +387,16 @@ mu_file_stream_create (mu_stream_t *pstream, const char *filename, int flags)
 int
 mu_fd_stream_create (mu_stream_t *pstream, char *filename, int fd, int flags)
 {
   struct _mu_file_stream *fstr;
   int rc = _mu_file_stream_create (&fstr,
 				   sizeof (struct _mu_file_stream),
-				   filename, fd, flags);
+				   filename, fd, flags|_MU_STR_OPEN);
   if (rc == 0)
     {
       mu_stream_t stream = (mu_stream_t) fstr;
       mu_stream_set_buffer (stream, mu_buffer_full, 0);
-      rc = mu_stream_open (stream);
-      if (rc)
-	mu_stream_unref (stream);
-      else
-	*pstream = stream;
+      *pstream = stream;
     }
   return rc;
 }
 
diff --git a/libmailutils/message_stream.c b/libmailutils/message_stream.c
index 6bfdae50a..9f590d9fd 100644
--- a/libmailutils/message_stream.c
+++ b/libmailutils/message_stream.c
@@ -411,14 +411,14 @@ mu_stream_to_message (mu_stream_t instream, mu_message_t *pmsg)
     {
       mu_message_destroy (&msg, draftstream);
       mu_stream_destroy (&draftstream);
       return rc;
     }
   
-  mu_envelope_set_date (env, _env_msg_date, msg);
-  mu_envelope_set_sender (env, _env_msg_sender, msg);
+  mu_envelope_set_date (env, _env_msg_date, draftstream);
+  mu_envelope_set_sender (env, _env_msg_sender, draftstream);
   mu_message_set_envelope (msg, env, draftstream);
 
   mu_body_create (&body, msg);
   /* FIXME: It would be cleaner to use ioctl here */
   sp = (struct _mu_message_stream *) draftstream;
   rc = mu_streamref_create_abridged (&bstream, instream,
diff --git a/libmailutils/stream.c b/libmailutils/stream.c
index 2dfbc82bc..608447907 100644
--- a/libmailutils/stream.c
+++ b/libmailutils/stream.c
@@ -31,12 +31,15 @@
 #include <mailutils/nls.h>
 #include <mailutils/stream.h>
 #include <mailutils/sys/stream.h>
 
 size_t mu_stream_default_buffer_size = MU_STREAM_DEFBUFSIZ;
 
+#define _MU_STR_FLUSH_ALL  0x01
+#define _MU_STR_FLUSH_KEEP 0x02
+
 #define _stream_event(stream, code, n, p)			\
   do								\
     {								\
       if ((stream)->event_cb &&					\
           ((stream)->event_mask & _MU_STR_EVMASK(code)))	\
         (stream)->event_cb (stream, code, n, p);		\
@@ -164,18 +167,18 @@ _stream_buffer_full_p (struct _mu_stream *stream)
 	return _stream_buffer_is_full (stream);
       }
     return 0;
 }
 
 static int
-_stream_flush_buffer (struct _mu_stream *stream, int all)
+_stream_flush_buffer (struct _mu_stream *stream, int flags)
 {
   int rc;
   char *start, *end;
   size_t wrsize;
-  
+
   if (stream->flags & _MU_STR_DIRTY)
     {
       if ((stream->flags & MU_STREAM_SEEK) && stream->seek)
 	{
 	  mu_off_t off;
 	  rc = stream->seek (stream, stream->offset, &off);
@@ -212,13 +215,14 @@ _stream_flush_buffer (struct _mu_stream *stream, int all)
 			     size, start);
 	      start += size;
 	      wrsize -= size;
 	      if (wrsize == 0)
 		break;
 	    }
-	  if ((all && wrsize) || wrsize == stream->level)
+	  if (((flags & _MU_STR_FLUSH_ALL) && wrsize) ||
+	      wrsize == stream->level)
 	    {
 	      rc = _stream_write_unbuffered (stream,
 					     stream->buffer,
 					     wrsize,
 					     1, NULL);
 	      if (rc)
@@ -231,16 +235,17 @@ _stream_flush_buffer (struct _mu_stream *stream, int all)
 	    memmove (stream->buffer, start, wrsize);
 	  else
 	    _stream_clrflag (stream, _MU_STR_DIRTY);
 	  stream->level = stream->pos = wrsize;
 	  return 0;
 	}
+      _stream_clrflag (stream, _MU_STR_DIRTY);
     }
 
-  _stream_clrflag (stream, _MU_STR_DIRTY);
-  stream->pos = stream->level = 0;
+  if (!(flags & _MU_STR_FLUSH_KEEP))
+    stream->pos = stream->level = 0;
   return 0;
 }
 
 
 mu_stream_t
 _mu_stream_create (size_t size, int flags)
@@ -397,38 +402,40 @@ mu_stream_seek (mu_stream_t stream, mu_off_t offset, int whence,
       offset += stream->offset;
       break;
 
     case MU_SEEK_END:
       rc = mu_stream_size (stream, &size);
       if (rc)
-	return mu_stream_seterr (stream, rc, 1);
-      offset += size + stream->pos;
+	return rc;
+      offset += size;
       break;
 
     default:
       return mu_stream_seterr (stream, EINVAL, 1);
     }
 
   if (stream->buftype == mu_buffer_none ?
       (offset != stream->offset)
        : (stream->level == 0
 	  || offset < stream->offset
 	  || offset > stream->offset + stream->level))
     {
-      if ((rc = _stream_flush_buffer (stream, 1)))
+      if ((rc = _stream_flush_buffer (stream, _MU_STR_FLUSH_ALL)))
 	return rc;
       rc = stream->seek (stream, offset, &stream->offset);
       if (rc == ESPIPE)
 	return rc;
       if (rc)
 	return mu_stream_seterr (stream, rc, 1);
       _mu_stream_cleareof (stream);
     }
   else if (stream->buftype != mu_buffer_none)
     stream->pos = offset - stream->offset;
-			    
+
+  _mu_stream_cleareof (stream);
+  
   if (pres)
     *pres = stream->offset + stream->pos;
   return 0;
 }
 
 /* Skip COUNT bytes from the current position in stream by reading from
@@ -466,13 +473,13 @@ _stream_skip_input_bytes (mu_stream_t stream, mu_off_t count, mu_off_t *pres)
 	    }
 	}
       else
 	{
 	  for (pos = 0;;)
 	    {
-	      if ((rc = _stream_flush_buffer (stream, 1)))
+	      if ((rc = _stream_flush_buffer (stream, _MU_STR_FLUSH_ALL)))
 		return rc;
 	      if (stream->pos == stream->level)
 		{
 		  rc = _stream_fill_buffer (stream);
 		  if (rc)
 		    break;
@@ -690,16 +697,21 @@ mu_stream_read (mu_stream_t stream, void *buf, size_t size, size_t *pread)
   if (stream->buftype == mu_buffer_none)
     return _stream_read_unbuffered (stream, buf, size, !pread, pread);
   else
     {
       char *bufp = buf;
       size_t nbytes = 0;
+      int rc;
+      
+      if ((rc = _stream_flush_buffer (stream,
+				      _MU_STR_FLUSH_ALL|_MU_STR_FLUSH_KEEP)))
+	return rc;
+
       while (size)
 	{
 	  size_t n;
-	  int rc;
 	  
 	  if (stream->pos == stream->level)
 	    {
 	      if ((rc = _stream_fill_buffer (stream)))
 		{
 		  if (nbytes)
@@ -821,13 +833,18 @@ mu_stream_readdelim (mu_stream_t stream, char *buf, size_t size,
 	  stream->offset += nread;
 	}
       else
 	rc = _stream_readdelim (stream, buf, size, delim, pread);
     }
   else
-    rc = _stream_scandelim (stream, buf, size, delim, pread);
+    {
+      if ((rc = _stream_flush_buffer (stream,
+				      _MU_STR_FLUSH_ALL|_MU_STR_FLUSH_KEEP)))
+	return rc;
+      rc = _stream_scandelim (stream, buf, size, delim, pread);
+    }
   return rc;
 }
 
 int
 mu_stream_readline (mu_stream_t stream, char *buf, size_t size, size_t *pread)
 {
@@ -847,12 +864,16 @@ mu_stream_getdelim (mu_stream_t stream, char **pbuf, size_t *psize,
     {
       if (stream->open)
 	return MU_ERR_NOT_OPEN;
       _stream_init (stream);
     }
 
+  if ((rc = _stream_flush_buffer (stream,
+				  _MU_STR_FLUSH_ALL|_MU_STR_FLUSH_KEEP)))
+    return rc;
+  
   if (lineptr == NULL || n == 0)
     {
       char *new_lineptr;
       n = 120;
       new_lineptr = mu_realloc (lineptr, n);
       if (new_lineptr == NULL) 
@@ -994,13 +1015,13 @@ mu_stream_flush (mu_stream_t stream)
   if (!(stream->flags & _MU_STR_OPEN))
     {
       if (stream->open)
 	return MU_ERR_NOT_OPEN;
       _stream_init (stream);
     }
-  rc = _stream_flush_buffer (stream, 1);
+  rc = _stream_flush_buffer (stream, _MU_STR_FLUSH_ALL);
   if (rc)
     return rc;
   if ((stream->flags & _MU_STR_WRT) && stream->flush)
     return stream->flush (stream);
   _stream_clrflag (stream, _MU_STR_WRT);
   return 0;
@@ -1028,22 +1049,29 @@ mu_stream_close (mu_stream_t stream)
 }
 
 int
 mu_stream_size (mu_stream_t stream, mu_off_t *psize)
 {
   int rc;
-    
+  mu_off_t size;
+  
   if (!(stream->flags & _MU_STR_OPEN))
     {
       if (stream->open)
 	return MU_ERR_NOT_OPEN;
       _stream_init (stream);
     }
   if (!stream->size)
     return mu_stream_seterr (stream, ENOSYS, 0);
-  rc = stream->size (stream, psize);
+  rc = stream->size (stream, &size);
+  if (rc == 0)
+    {
+      if (stream->buftype != mu_buffer_none && stream->offset == size)
+	size += stream->level;
+      *psize = size;
+    }
   return mu_stream_seterr (stream, rc, rc != 0);
 }
 
 mu_off_t
 mu_stream_bytes_in (mu_stream_t stream)
 {
@@ -1115,13 +1143,13 @@ mu_stream_truncate (mu_stream_t stream, mu_off_t size)
     }
   
   if (stream->truncate)
     {
       int rc;
       
-      if ((rc = _stream_flush_buffer (stream, 1)))
+      if ((rc = _stream_flush_buffer (stream, _MU_STR_FLUSH_ALL)))
 	return rc;
       return stream->truncate (stream, size);
     }
   return ENOSYS;
 }
 
diff --git a/libmailutils/streamcpy.c b/libmailutils/streamcpy.c
index 13742e79a..2020fad2c 100644
--- a/libmailutils/streamcpy.c
+++ b/libmailutils/streamcpy.c
@@ -69,12 +69,13 @@ mu_stream_copy (mu_stream_t dst, mu_stream_t src, mu_off_t size,
 	      size -= pos;
 	      break;
       
 	    case EACCES:
 	      mu_stream_clearerr (src);
 	    case ENOSYS:
+	      size = 0;
 	      break;
 
 	    default:
 	      return status;
 	    }
 	}
diff --git a/libmailutils/temp_file_stream.c b/libmailutils/temp_file_stream.c
index 857bada0c..fb178a5dc 100644
--- a/libmailutils/temp_file_stream.c
+++ b/libmailutils/temp_file_stream.c
@@ -63,10 +63,13 @@ mu_temp_file_stream_create (mu_stream_t *pstream, const char *dir)
       str->flags = _MU_FILE_STREAM_TEMP;
       stream = (mu_stream_t) str;
       rc = mu_stream_open (stream);
       if (rc)
 	mu_stream_unref (stream);
       else
-	*pstream = stream;
+	{
+	  mu_stream_set_buffer (stream, mu_buffer_full, 0);
+	  *pstream = stream;
+	}
     }
   return 0;
 }
diff --git a/maidag/Makefile.am b/maidag/Makefile.am
index 0b28d9373..835f8953a 100644
--- a/maidag/Makefile.am
+++ b/maidag/Makefile.am
@@ -23,13 +23,12 @@ maidag_SOURCES=\
  deliver.c\
  forward.c\
  guile.c\
  lmtp.c\
  maidag.c\
  maidag.h\
- mailtmp.c\
  mailquota.c\
  python.c\
  sieve.c\
  script.c\
  util.c
 
diff --git a/maidag/deliver.c b/maidag/deliver.c
index 24e2f0cbb..d5bbc1f53 100644
--- a/maidag/deliver.c
+++ b/maidag/deliver.c
@@ -16,32 +16,116 @@
    along with GNU Mailutils; if not, write to the Free Software
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
    MA 02110-1301 USA */
 
 #include "maidag.h"
 
-void
-make_tmp (const char *from, mu_mailbox_t *mbox)
+static mu_mailbox_t
+make_tmp (const char *from)
 {
-  struct mail_tmp *mtmp;
-  char *buf = NULL;
-  size_t n = 0;
   int rc;
+  mu_stream_t in, out;
+  char *buf = NULL;
+  size_t size = 0, n;
+  mu_mailbox_t mbox;
+  
+  rc = mu_stdio_stream_create (&in, MU_STDIN_FD, MU_STREAM_READ);
+  if (rc)
+    {
+      mu_diag_funcall (MU_DIAG_ERROR, "mu_stdio_stream_create",
+		       "MU_STDIN_FD", rc);
+      exit (EX_TEMPFAIL);
+    } 
+
+  rc = mu_temp_file_stream_create (&out, NULL);
+  if (rc)
+    {
+      maidag_error (_("unable to open temporary file: %s"), mu_strerror (rc));
+      exit (EX_TEMPFAIL);
+    }
 
-  if (mail_tmp_begin (&mtmp, from))
-    exit (EX_TEMPFAIL);
+  rc = mu_stream_getline (in, &buf, &size, &n);
+  if (rc)
+    {
+      maidag_error (_("read error: %s"), mu_strerror (rc));
+      mu_stream_destroy (&in);
+      mu_stream_destroy (&out);
+      exit (EX_TEMPFAIL);
+    }
+  if (n == 0)
+    {
+      maidag_error (_("unexpected EOF on input"));
+      mu_stream_destroy (&in);
+      mu_stream_destroy (&out);
+      exit (EX_TEMPFAIL);
+    } 
 
-  while (getline (&buf, &n, stdin) > 0)
-    if ((rc = mail_tmp_add_line (mtmp, buf, strlen (buf))))
-      break;
+  if (n >= 5 && memcmp (buf, "From ", 5))
+    {
+      struct mu_auth_data *auth = NULL;
+      if (!from)
+	{
+	  auth = mu_get_auth_by_uid (getuid ());
+	  if (auth)
+	    from = auth->name;
+	}
+      if (from)
+	{
+	  time_t t;
+	  
+	  time (&t);
+	  mu_stream_printf (out, "From %s %s", from, ctime (&t));
+	}
+      else
+	{
+	  maidag_error (_("cannot determine sender address"));
+	  mu_stream_destroy (&in);
+	  mu_stream_destroy (&out);
+	  exit (EX_TEMPFAIL);
+	}
+      if (auth)
+	mu_auth_data_free (auth);
+    }
+
+  mu_stream_write (out, buf, n, NULL);
   free (buf);
-  if (rc == 0)
-    rc = mail_tmp_finish (mtmp, mbox);
-  mail_tmp_destroy (&mtmp);
+  
+  rc = mu_stream_copy (out, in, 0, NULL);
+  
+  mu_stream_destroy (&in);
   if (rc)
-    exit (EX_TEMPFAIL);
+    {
+      maidag_error (_("copy error: %s"), mu_strerror (rc));
+      mu_stream_destroy (&out);
+      exit (EX_TEMPFAIL);
+    }
+
+  mu_stream_flush (out);
+  if ((rc = mu_mailbox_create (&mbox, "mbox:/dev/null")) 
+      || (rc = mu_mailbox_open (mbox, MU_STREAM_READ))
+      || (rc = mu_mailbox_set_stream (mbox, out)))
+    {
+      maidag_error (_("error opening temporary file: %s"), 
+                    mu_strerror (rc));
+      mu_stream_destroy (&out);
+      exit (EX_TEMPFAIL);
+    }
+
+  rc = mu_mailbox_messages_count (mbox, &n);
+  if (rc)
+    {
+      errno = rc;
+      maidag_error (_("error creating temporary message: %s"),
+		    mu_strerror (rc));
+      mu_stream_destroy (&out);
+      exit (EX_TEMPFAIL);
+    }
+
+  /* FIXME: mu_stream_unref (out); But mu_mailbox_set_stream
+     steals the reference */
+  return mbox;
 }
 
 int
 mda (mu_mailbox_t mbx, char *username)
 {
   int status;
@@ -62,15 +146,13 @@ mda (mu_mailbox_t mbx, char *username)
   return exit_code;
 }
 
 int
 maidag_stdio_delivery (int argc, char **argv)
 {
-  mu_mailbox_t mbox;
-
-  make_tmp (sender_address, &mbox);
+  mu_mailbox_t mbox = make_tmp (sender_address);
   
   if (multiple_delivery)
     multiple_delivery = argc > 1;
 
   for (; *argv; argv++)
     mda (mbox, *argv);
diff --git a/maidag/lmtp.c b/maidag/lmtp.c
index 439329112..726be1530 100644
--- a/maidag/lmtp.c
+++ b/maidag/lmtp.c
@@ -23,32 +23,53 @@
 #include <sys/un.h>
 #include <sys/wait.h>
 #include <signal.h>
 #include <mu_umaxtostr.h>
 
 mu_list_t lmtp_groups;
-static int lmtp_transcript;
+
+static mu_stream_t
+lmpt_transcript (mu_stream_t iostream)
+{
+  int rc;
+  mu_debug_t debug;
+  mu_stream_t dstr, xstr;
+      
+  mu_diag_get_debug (&debug);
+      
+  rc = mu_dbgstream_create (&dstr, debug, MU_DIAG_DEBUG, 0);
+  if (rc)
+    mu_error (_("cannot create debug stream; transcript disabled: %s"),
+	      mu_strerror (rc));
+  else
+    {
+      rc = mu_xscript_stream_create (&xstr, iostream, dstr, NULL);
+      if (rc)
+	mu_error (_("cannot create transcript stream: %s"),
+		  mu_strerror (rc));
+      else
+	{
+	  /* FIXME: Would do a mu_stream_unref (iostream) here,
+	     however mu_xscript_stream_create *may* steal the reference.
+	     This should be fixed in mu_xscript_stream_create. */
+	  iostream = xstr;
+	}
+    }
+  return iostream;
+}
 
 void
-lmtp_reply (FILE *fp, char *code, char *enh, char *fmt, ...)
+lmtp_reply (mu_stream_t iostr, char *code, char *enh, char *fmt, ...)
 {
   va_list ap;
   char *str;
   
   va_start (ap, fmt);
   vasprintf (&str, fmt, ap);
   va_end (ap);
 
-  if (lmtp_transcript)
-    {
-      if (enh)
-	mu_diag_output (MU_DIAG_INFO, "LMTP reply: %s %s %s", code, enh, str);
-      else
-	mu_diag_output (MU_DIAG_INFO, "LMTP reply: %s %s", code, str);
-    }
-  
   if (!str)
     {
       mu_error (_("not enough memory"));
       exit (EX_TEMPFAIL);
     }
 
@@ -56,45 +77,29 @@ lmtp_reply (FILE *fp, char *code, char *enh, char *fmt, ...)
     {
       char *end = strchr (str, '\n');
 
       if (end)
 	{
 	  size_t len = end - str;
-	  fprintf (fp, "%s-", code);
+	  mu_stream_printf (iostr, "%s-", code);
 	  if (enh)
-	    fprintf (fp, "%s ", enh);
-	  fprintf (fp, "%.*s\r\n", (int) len, str);
+	    mu_stream_printf (iostr, "%s ", enh);
+	  mu_stream_printf (iostr, "%.*s\r\n", (int) len, str);
 	  for (str = end; *str && *str == '\n'; str++);
 	}
       else
 	{
-	  fprintf (fp, "%s ", code);
+	  mu_stream_printf (iostr, "%s ", code);
 	  if (enh)
-	    fprintf (fp, "%s ", enh);
-	  fprintf (fp, "%s\r\n", str);
+	    mu_stream_printf (iostr, "%s ", enh);
+	  mu_stream_printf (iostr, "%s\r\n", str);
 	  str += strlen (str);
 	}
     }
 }
 
-void
-xlatnl (char *arg)
-{
-  size_t len = strlen (arg);
-  if (len > 0 && arg[len-1] == '\n')
-    {
-      len--;
-      if (len > 0 && arg[len-1] == '\r')
-	{
-	  arg[len-1] = '\n';
-	  arg[len] = 0;
-	}
-    }
-}
-
-
 enum lmtp_state
   {
     state_none,
     
     state_init,    
     state_lhlo,    
@@ -149,20 +154,19 @@ int transtab[NCMD][NSTATE] = {
 
 
 /* Delivery data */
 char *lhlo_domain;     /* Sender domain */
 char *mail_from;       /* Sender address */
 mu_list_t rcpt_list;   /* Recipient addresses */
-struct mail_tmp *mtmp; /* Temporary mail storage */
-mu_mailbox_t mbox;     /* Collected mail body */
+mu_message_t mesg;     /* Collected message */
 
 
 int
-cfun_unknown (FILE *out, char *arg)
+cfun_unknown (mu_stream_t iostr, char *arg)
 {
-  lmtp_reply (out, "500", "5.5.1", "Command unrecognized");
+  lmtp_reply (iostr, "500", "5.5.1", "Command unrecognized");
   return 0;
 }
 
 
 static void
 add_default_domain (char *str, int len, char **pret)
@@ -239,229 +243,262 @@ check_address (char *arg, int with_domain, char **pret)
     }
   return 0;
 }
 
 
 int
-cfun_mail_from (FILE *out, char *arg)
+cfun_mail_from (mu_stream_t iostr, char *arg)
 {
   if (*arg == 0)
     {
-      lmtp_reply (out, "501", "5.5.2", "Syntax error");
+      lmtp_reply (iostr, "501", "5.5.2", "Syntax error");
       return 1;
     }
 
   if (check_address (arg, 1, &mail_from))
     {
-      lmtp_reply (out, "553", "5.1.8", "Address format error");
+      lmtp_reply (iostr, "553", "5.1.8", "Address format error");
       return 1;
     }
-  lmtp_reply (out, "250", "2.1.0", "Go ahead");
+  lmtp_reply (iostr, "250", "2.1.0", "Go ahead");
   return 0;
 }
 
 
 int
-cfun_rcpt_to (FILE *out, char *arg)
+cfun_rcpt_to (mu_stream_t iostr, char *arg)
 {
   char *user;
   struct mu_auth_data *auth;
   
   if (*arg == 0)
     {
-      lmtp_reply (out, "501", "5.5.2", "Syntax error");
+      lmtp_reply (iostr, "501", "5.5.2", "Syntax error");
       return 1;
     }
 
   /* FIXME: Check if domain is OK */
   if (check_address (arg, 0, &user))
     {
-      lmtp_reply (out, "553", "5.1.8", "Address format error");
+      lmtp_reply (iostr, "553", "5.1.8", "Address format error");
       return 1;
     }
   auth = mu_get_auth_by_name (user);
   if (!auth)
     {
-      lmtp_reply (out, "550", "5.1.1", "User unknown");
+      lmtp_reply (iostr, "550", "5.1.1", "User unknown");
       free (user);
       return 1;
     }
   mu_auth_data_free (auth);
   if (!rcpt_list)
     {
       mu_list_create (&rcpt_list);
       mu_list_set_destroy_item (rcpt_list, mu_list_free_item);
     }
   mu_list_append (rcpt_list, user);
-  lmtp_reply (out, "250", "2.1.5", "Go ahead");
+  lmtp_reply (iostr, "250", "2.1.5", "Go ahead");
   return 0;
 }  
 
 
 int
-cfun_data (FILE *out, char *arg)
-{
-  if (*arg)
-    {
-      lmtp_reply (out, "501", "5.5.2", "Syntax error");
-      return 1;
-    }
-
-  if (mail_tmp_begin (&mtmp, mail_from))
-    {
-      /* FIXME: codes */
-      lmtp_reply (out, "450", "4.1.0", "Temporary failure, try again later");
-      return 1;
-    }
-  lmtp_reply (out, "354", NULL, "Go ahead");
-  return 0;
-}
-
-
-int
 dot_temp_fail (void *item, void *cbdata)
 {
   char *name = item;
-  FILE *out = cbdata;
-  lmtp_reply (out, "450", "4.1.0", "%s: temporary failure", name);
+  mu_stream_t iostr = cbdata;
+  lmtp_reply (iostr, "450", "4.1.0", "%s: temporary failure", name);
   return 0;
 }
 
 int
 dot_deliver (void *item, void *cbdata)
 {
   char *name = item;
-  FILE *out = cbdata;
+  mu_stream_t iostr = cbdata;
   char *errp = NULL;
-  mu_message_t msg;
-  int status;
   
-  if ((status = mu_mailbox_get_message (mbox, 1, &msg)) != 0)
-    {
-      mu_error (_("cannot get message from the temporary mailbox: %s"),
-		mu_strerror (status));
-      lmtp_reply (out, "450", "4.1.0",
-		  "%s: temporary failure, try again later",
-		  name);
-      return 0;
-    }
-
-  switch (deliver (msg, name, &errp))
+  switch (deliver (mesg, name, &errp))
     {
     case 0:
-      lmtp_reply (out, "250", "2.0.0", "%s: delivered", name);
+      lmtp_reply (iostr, "250", "2.0.0", "%s: delivered", name);
       break;
 
     case EX_UNAVAILABLE:
       if (errp)
-	lmtp_reply (out, "553", "5.1.8", "%s", errp);
+	lmtp_reply (iostr, "553", "5.1.8", "%s", errp);
       else
-	lmtp_reply (out, "553", "5.1.8", "%s: delivery failed", name);
+	lmtp_reply (iostr, "553", "5.1.8", "%s: delivery failed", name);
       break;
 
     default:
       if (errp)
-	lmtp_reply (out, "450", "4.1.0", "%s", errp);
+	lmtp_reply (iostr, "450", "4.1.0", "%s", errp);
       else
-	lmtp_reply (out, "450", "4.1.0",
+	lmtp_reply (iostr, "450", "4.1.0",
 		    "%s: temporary failure, try again later",
 		    name);
       break;
     }
   free (errp);
   return 0;
 }
 
 int
-cfun_dot (FILE *out, char *arg)
+cfun_data (mu_stream_t iostr, char *arg)
 {
-  if (!mtmp)
-    mu_list_do (rcpt_list, dot_temp_fail, out);
-  else
+  int rc;
+  mu_stream_t flt, tempstr;
+  time_t t;
+  int xlev = MU_XSCRIPT_PAYLOAD, xlev_switch = 0, buf_switch = 0;
+  struct mu_buffer_query oldbuf;
+  
+  if (*arg)
     {
-      int rc = mail_tmp_finish (mtmp, &mbox);
-      if (rc)
-	mu_list_do (rcpt_list, dot_temp_fail, out);
-      else
-	{
-	  mu_list_do (rcpt_list, dot_deliver, out);
-	  mail_tmp_destroy (&mtmp);
-	  mu_mailbox_destroy (&mbox);
-	}
+      lmtp_reply (iostr, "501", "5.5.2", "Syntax error");
+      return 1;
     }
-  free (mail_from);
-  mu_list_destroy (&rcpt_list);
+
+  rc = mu_filter_create (&flt, iostr, "CRLFDOT", MU_FILTER_DECODE,
+			 MU_STREAM_READ|MU_STREAM_WRTHRU);
+  if (rc)
+    {
+      maidag_error (_("unable to open filter: %s"),
+		    mu_strerror (rc));
+      lmtp_reply (iostr, "450", "4.1.0", "Temporary failure, try again later");
+      return 1;
+    }
+
+  rc = mu_temp_file_stream_create (&tempstr, NULL);
+  if (rc)
+    {
+      maidag_error (_("unable to open temporary file: %s"), mu_strerror (rc));
+      mu_stream_destroy (&flt);
+      return 1;
+    }
+
+  /* Write out envelope */
+  time (&t);
+  rc = mu_stream_printf (tempstr, "From %s %s", mail_from, ctime (&t));
+  if (rc)
+    {
+      maidag_error (_("copy error: %s"), mu_strerror (rc));
+      mu_stream_destroy (&flt);
+      mu_stream_destroy (&tempstr);
+      mu_list_do (rcpt_list, dot_temp_fail, iostr);
+    }
+
+  lmtp_reply (iostr, "354", NULL, "Go ahead");
+
+  if (mu_stream_ioctl (iostr, MU_IOCTL_GET_TRANSPORT_BUFFER, &oldbuf) == 0)
+    {
+      struct mu_buffer_query newbuf;
+
+      newbuf.type = MU_TRANSPORT_OUTPUT;
+      newbuf.buftype = mu_buffer_full;
+      newbuf.bufsize = 64*1024;
+      if (mu_stream_ioctl (iostr, MU_IOCTL_SET_TRANSPORT_BUFFER, &newbuf))
+	buf_switch = 1;
+    }
+
+  if (mu_stream_ioctl (iostr, MU_IOCTL_LEVEL, &xlev) == 0)
+    xlev_switch = 1;
+  rc = mu_stream_copy (tempstr, flt, 0, NULL);
+  mu_stream_destroy (&flt);
+  if (xlev_switch)
+    mu_stream_ioctl (iostr, MU_IOCTL_LEVEL, &xlev);
+  if (buf_switch)
+    mu_stream_ioctl (iostr, MU_IOCTL_SET_TRANSPORT_BUFFER, &oldbuf);
+  if (rc)
+    {
+      maidag_error (_("copy error: %s"), mu_strerror (rc));
+      mu_list_do (rcpt_list, dot_temp_fail, iostr);
+    }
+
+  rc = mu_stream_to_message (tempstr, &mesg);
+  if (rc)
+    {
+      maidag_error (_("error creating temporary message: %s"),
+		    mu_strerror (rc));
+      mu_list_do (rcpt_list, dot_temp_fail, iostr);
+    }
+  
+  rc = mu_list_do (rcpt_list, dot_deliver, iostr);
+
+  mu_stream_destroy (&tempstr);
+  mu_message_destroy (&mesg, mu_message_get_owner (mesg));
+  if (rc)
+    mu_list_do (rcpt_list, dot_temp_fail, iostr);
+
   return 0;
 }
-  
+
 
 int
-cfun_rset (FILE *out, char *arg)
+cfun_rset (mu_stream_t iostr, char *arg)
 {
   free (lhlo_domain);
   free (mail_from);
   mu_list_destroy (&rcpt_list);
-  mail_tmp_destroy (&mtmp);
-  mu_mailbox_destroy (&mbox);
-  lmtp_reply (out, "250", "2.0.0", "OK, forgotten");
+  mu_message_destroy (&mesg, mu_message_get_owner (mesg));
+  lmtp_reply (iostr, "250", "2.0.0", "OK, forgotten");
   return 0;
 }
 
 
 char *capa_str = "ENHANCEDSTATUSCODES\n\
 PIPELINING\n\
 8BITMIME\n\
 HELP";
 
 int
-cfun_lhlo (FILE *out, char *arg)
+cfun_lhlo (mu_stream_t iostr, char *arg)
 {
   if (*arg == 0)
     {
-      lmtp_reply (out, "501", "5.0.0", "Syntax error");
+      lmtp_reply (iostr, "501", "5.0.0", "Syntax error");
       return 1;
     }
   lhlo_domain = strdup (arg);
-  lmtp_reply (out, "250", NULL, "Hello\n");
-  lmtp_reply (out, "250", NULL, capa_str);
+  lmtp_reply (iostr, "250", NULL, "Hello\n");
+  lmtp_reply (iostr, "250", NULL, capa_str);
   return 0;
 }
 
 
 int
-cfun_quit (FILE *out, char *arg)
+cfun_quit (mu_stream_t iostr, char *arg)
 {
-  lmtp_reply (out, "221", "2.0.0", "Bye");
+  lmtp_reply (iostr, "221", "2.0.0", "Bye");
   return 0;
 }
 
 
 int
-cfun_help (FILE *out, char *arg)
+cfun_help (mu_stream_t iostr, char *arg)
 {
-  lmtp_reply (out, "200", "2.0.0", "Man, help yourself");
+  lmtp_reply (iostr, "200", "2.0.0", "Man, help yourself");
   return 0;
 }
 
 
 struct command_tab
 {
   char *cmd_verb;
   int cmd_len;
   enum lmtp_command cmd_code;
-  int (*cmd_fun) (FILE *, char *);
+  int (*cmd_fun) (mu_stream_t, char *);
 } command_tab[] = {
 #define S(s) #s, (sizeof #s - 1)
   { S(lhlo), cmd_lhlo, cfun_lhlo },
   { S(mail from:), cmd_mail, cfun_mail_from },   
   { S(rcpt to:), cmd_rcpt, cfun_rcpt_to },   
   { S(data), cmd_data, cfun_data },
   { S(quit), cmd_quit, cfun_quit },
   { S(rset), cmd_rset, cfun_rset },
   { S(help), cmd_help, cfun_help },
-  { S(.), cmd_dot, cfun_dot },
   { NULL, 0, cmd_unknown, cfun_unknown }
 };
 
 struct command_tab *
 getcmd (char *buf, char **sp)
 {
@@ -476,78 +513,53 @@ getcmd (char *buf, char **sp)
 	  return cp;
 	}
     }
   return cp;
 }
 
-static char *
-to_fgets (char *buf, size_t size, FILE *fp, unsigned int timeout)
+static int
+to_fgets (mu_stream_t iostr, char **pbuf, size_t *psize, size_t *pnread,
+	  unsigned int timeout)
 {
-  char *p;
+  int rc;
+  
   alarm (timeout);
-  p = fgets (buf, size, fp);
+  rc = mu_stream_getline (iostr, pbuf, psize, pnread);
   alarm (0);
-  return p;
+  return rc;
 }
 
 int
-lmtp_loop (FILE *in, FILE *out, unsigned int timeout)
+lmtp_loop (mu_stream_t iostr, unsigned int timeout)
 {
-  char buf[1024];
+  size_t size = 0, n;
+  char *buf = NULL;
   enum lmtp_state state = state_init;
 
-  setvbuf (in, NULL, _IOLBF, 0);
-  setvbuf (out, NULL, _IOLBF, 0);
-
-  lmtp_reply (out, "220", NULL, "At your service");
-  while (to_fgets (buf, sizeof buf, in, timeout))
+  lmtp_reply (iostr, "220", NULL, "At your service");
+  while (to_fgets (iostr, &buf, &size, &n, timeout) == 0 && n)
     {
-      if (state == state_data
-	  && !(buf[0] == '.'
-	       && (buf[1] == '\n' || (buf[1] == '\r' && buf[2] == '\n'))))
-	{
-	  /* This is a special case */
-	  if (mtmp)
-	    {
-	      size_t len;
-	      int rc;
+      char *sp;
+      struct command_tab *cp = getcmd (buf, &sp);
+      enum lmtp_command cmd = cp->cmd_code;
+      enum lmtp_state next_state = transtab[cmd][state];
 
-	      xlatnl (buf);