pass advertised facility to continued-transport process
[exim.git] / src / src / transport.c
index 54252575440f0ed5e2e6137e7b58bc8b9cad6b33..88d925e39904fd2c98a71345e397f0648ae24fde 100644 (file)
@@ -1,10 +1,8 @@
-/* $Cambridge: exim/src/src/transport.c,v 1.3 2005/01/04 10:00:42 ph10 Exp $ */
-
 /*************************************************
 *     Exim - an Internet mail transport agent    *
 *************************************************/
 
-/* Copyright (c) University of Cambridge 1995 - 2005 */
+/* Copyright (c) University of Cambridge 1995 - 2016 */
 /* See the file NOTICE for conditions of use and distribution. */
 
 /* General functions concerned with transportation, and generic options for all
@@ -13,6 +11,9 @@ transports. */
 
 #include "exim.h"
 
+#ifdef HAVE_LINUX_SENDFILE
+#include <sys/sendfile.h>
+#endif
 
 /* Structure for keeping list of addresses that have been added to
 Envelope-To:, in order to avoid duplication. */
@@ -65,13 +66,17 @@ optionlist optionlist_transports[] = {
                  (void *)offsetof(transport_instance, driver_name) },
   { "envelope_to_add",   opt_bool|opt_public,
                  (void *)(offsetof(transport_instance, envelope_to_add)) },
+#ifndef DISABLE_EVENT
+  { "event_action",     opt_stringptr | opt_public,
+                 (void *)offsetof(transport_instance, event_action) },
+#endif
   { "group",             opt_expand_gid|opt_public,
                  (void *)offsetof(transport_instance, gid) },
-  { "headers_add",      opt_stringptr|opt_public,
+  { "headers_add",      opt_stringptr|opt_public|opt_rep_str,
                  (void *)offsetof(transport_instance, add_headers) },
   { "headers_only",     opt_bool|opt_public,
                  (void *)offsetof(transport_instance, headers_only) },
-  { "headers_remove",   opt_stringptr|opt_public,
+  { "headers_remove",   opt_stringptr|opt_public|opt_rep_str,
                  (void *)offsetof(transport_instance, remove_headers) },
   { "headers_rewrite",  opt_rewrite|opt_public,
                  (void *)offsetof(transport_instance, headers_rewrite) },
@@ -79,6 +84,8 @@ optionlist optionlist_transports[] = {
                  (void *)offsetof(transport_instance, home_dir) },
   { "initgroups",       opt_bool|opt_public,
                  (void *)offsetof(transport_instance, initgroups) },
+  { "max_parallel",     opt_stringptr|opt_public,
+                 (void *)offsetof(transport_instance, max_parallel) },
   { "message_size_limit", opt_stringptr|opt_public,
                  (void *)offsetof(transport_instance, message_size_limit) },
   { "rcpt_include_affixes", opt_bool|opt_public,
@@ -199,26 +206,44 @@ BOOL
 transport_write_block(int fd, uschar *block, int len)
 {
 int i, rc, save_errno;
+int local_timeout = transport_write_timeout;
+
+/* This loop is for handling incomplete writes and other retries. In most
+normal cases, it is only ever executed once. */
 
 for (i = 0; i < 100; i++)
   {
   DEBUG(D_transport)
     debug_printf("writing data block fd=%d size=%d timeout=%d\n",
-      fd, len, transport_write_timeout);
-  if (transport_write_timeout > 0) alarm(transport_write_timeout);
+      fd, len, local_timeout);
 
-  #ifdef SUPPORT_TLS
-  if (tls_active == fd) rc = tls_write(block, len); else
-  #endif
+  /* This code makes use of alarm() in order to implement the timeout. This
+  isn't a very tidy way of doing things. Using non-blocking I/O with select()
+  provides a neater approach. However, I don't know how to do this when TLS is
+  in use. */
 
-  rc = write(fd, block, len);
-  save_errno = errno;
+  if (transport_write_timeout <= 0)   /* No timeout wanted */
+    {
+    #ifdef SUPPORT_TLS
+    if (tls_out.active == fd) rc = tls_write(FALSE, block, len); else
+    #endif
+    rc = write(fd, block, len);
+    save_errno = errno;
+    }
 
-  /* Cancel the alarm and deal with a timeout */
+  /* Timeout wanted. */
 
-  if (transport_write_timeout > 0)
+  else
     {
-    alarm(0);
+    alarm(local_timeout);
+#ifdef SUPPORT_TLS
+    if (tls_out.active == fd)
+      rc = tls_write(FALSE, block, len);
+    else
+#endif
+      rc = write(fd, block, len);
+    save_errno = errno;
+    local_timeout = alarm(0);
     if (sigalrm_seen)
       {
       errno = ETIMEDOUT;
@@ -230,7 +255,8 @@ for (i = 0; i < 100; i++)
 
   if (rc == len) { transport_count += len; return TRUE; }
 
-  /* A non-negative return code is an incomplete write. Try again. */
+  /* A non-negative return code is an incomplete write. Try again for the rest
+  of the block. If we have exactly hit the timeout, give up. */
 
   if (rc >= 0)
     {
@@ -238,7 +264,7 @@ for (i = 0; i < 100; i++)
     block += rc;
     transport_count += rc;
     DEBUG(D_transport) debug_printf("write incomplete (%d)\n", rc);
-    continue;
+    goto CHECK_TIMEOUT;   /* A few lines below */
     }
 
   /* A negative return code with an EINTR error is another form of
@@ -248,7 +274,7 @@ for (i = 0; i < 100; i++)
     {
     DEBUG(D_transport)
       debug_printf("write interrupted before anything written\n");
-    continue;
+    goto CHECK_TIMEOUT;   /* A few lines below */
     }
 
   /* A response of EAGAIN from write() is likely only in the case of writing
@@ -259,6 +285,16 @@ for (i = 0; i < 100; i++)
     DEBUG(D_transport)
       debug_printf("write temporarily locked out, waiting 1 sec\n");
     sleep(1);
+
+    /* Before continuing to try another write, check that we haven't run out of
+    time. */
+
+    CHECK_TIMEOUT:
+    if (transport_write_timeout > 0 && local_timeout <= 0)
+      {
+      errno = ETIMEDOUT;
+      return FALSE;
+      }
     continue;
     }
 
@@ -294,7 +330,7 @@ Returns:      the yield of transport_write_block()
 */
 
 BOOL
-transport_write_string(int fd, char *format, ...)
+transport_write_string(int fd, const char *format, ...)
 {
 va_list ap;
 va_start(ap, format);
@@ -323,7 +359,7 @@ Arguments:
   fd         file descript to write to
   chunk      pointer to data to write
   len        length of data to write
-  usr_crlf   TRUE if CR LF is wanted at the end of each line
+  tctx       transport context - processing to be done during output
 
 In addition, the static nl_xxx variables must be set as required.
 
@@ -331,11 +367,11 @@ Returns:     TRUE on success, FALSE on failure (with errno preserved)
 */
 
 static BOOL
-write_chunk(int fd, uschar *chunk, int len, BOOL use_crlf)
+write_chunk(int fd, transport_ctx * tctx, uschar *chunk, int len)
 {
 uschar *start = chunk;
 uschar *end = chunk + len;
-register uschar *ptr;
+uschar *ptr;
 int mlen = DELIVER_OUT_BUFFER_SIZE - nl_escape_length - 2;
 
 /* The assumption is made that the check string will never stretch over move
@@ -374,16 +410,22 @@ possible. */
 
 for (ptr = start; ptr < end; ptr++)
   {
-  register int ch;
+  int ch, len;
 
   /* Flush the buffer if it has reached the threshold - we want to leave enough
   room for the next uschar, plus a possible extra CR for an LF, plus the escape
   string. */
 
-  if (chunk_ptr - deliver_out_buffer > mlen)
+  if ((len = chunk_ptr - deliver_out_buffer) > mlen)
     {
-    if (!transport_write_block(fd, deliver_out_buffer,
-          chunk_ptr - deliver_out_buffer))
+    /* If CHUNKING, prefix with BDAT (size) NON-LAST.  Also, reap responses
+    from previous SMTP commands. */
+
+    if (tctx &&  tctx->options & topt_use_bdat  &&  tctx->chunk_cb)
+      if (tctx->chunk_cb(fd, tctx, (unsigned)len, tc_reap_prev|tc_reap_one) != OK)
+       return FALSE;
+
+    if (!transport_write_block(fd, deliver_out_buffer, len))
       return FALSE;
     chunk_ptr = deliver_out_buffer;
     }
@@ -394,8 +436,9 @@ for (ptr = start; ptr < end; ptr++)
 
     /* Insert CR before NL if required */
 
-    if (use_crlf) *chunk_ptr++ = '\r';
+    if (tctx  &&  tctx->options & topt_use_crlf) *chunk_ptr++ = '\r';
     *chunk_ptr++ = '\n';
+    transport_newlines++;
 
     /* The check_string test (formerly "from hack") replaces the specific
     string at the start of a line with an escape string (e.g. "From " becomes
@@ -510,14 +553,14 @@ Arguments:
   pdlist    address of anchor of the list of processed addresses
   first     TRUE if this is the first address; set it FALSE afterwards
   fd        the file descriptor to write to
-  use_crlf  to be passed on to write_chunk()
+  tctx      transport context - processing to be done during output
 
 Returns:    FALSE if writing failed
 */
 
 static BOOL
 write_env_to(address_item *p, struct aci **pplist, struct aci **pdlist,
-  BOOL *first, int fd, BOOL use_crlf)
+  BOOL *first, int fd, transport_ctx * tctx)
 {
 address_item *pp;
 struct aci *ppp;
@@ -525,8 +568,7 @@ struct aci *ppp;
 /* Do nothing if we have already handled this address. If not, remember it
 so that we don't handle it again. */
 
-for (ppp = *pdlist; ppp != NULL; ppp = ppp->next)
-  { if (p == ppp->ptr) return TRUE; }
+for (ppp = *pdlist; ppp; ppp = ppp->next) if (p == ppp->ptr) return TRUE;
 
 ppp = store_get(sizeof(struct aci));
 ppp->next = *pdlist;
@@ -538,19 +580,17 @@ ppp->ptr = p;
 for (pp = p;; pp = pp->parent)
   {
   address_item *dup;
-  for (dup = addr_duplicate; dup != NULL; dup = dup->next)
-    {
-    if (dup->dupof != pp) continue;   /* Not a dup of our address */
-    if (!write_env_to(dup, pplist, pdlist, first, fd, use_crlf)) return FALSE;
-    }
-  if (pp->parent == NULL) break;
+  for (dup = addr_duplicate; dup; dup = dup->next)
+    if (dup->dupof == pp)   /* a dup of our address */
+      if (!write_env_to(dup, pplist, pdlist, first, fd, tctx))
+       return FALSE;
+  if (!pp->parent) break;
   }
 
 /* Check to see if we have already output the progenitor. */
 
-for (ppp = *pplist; ppp != NULL; ppp = ppp->next)
-  { if (pp == ppp->ptr) break; }
-if (ppp != NULL) return TRUE;
+for (ppp = *pplist; ppp; ppp = ppp->next) if (pp == ppp->ptr) break;
+if (ppp) return TRUE;
 
 /* Remember what we have output, and output it. */
 
@@ -559,14 +599,184 @@ ppp->next = *pplist;
 *pplist = ppp;
 ppp->ptr = pp;
 
-if (!(*first) && !write_chunk(fd, US",\n ", 3, use_crlf)) return FALSE;
+if (!*first && !write_chunk(fd, tctx, US",\n ", 3)) return FALSE;
 *first = FALSE;
-return write_chunk(fd, pp->address, Ustrlen(pp->address), use_crlf);
+return write_chunk(fd, tctx, pp->address, Ustrlen(pp->address));
 }
 
 
 
 
+/* Add/remove/rewwrite headers, and send them plus the empty-line sparator.
+
+Globals:
+  header_list
+
+Arguments:
+  addr                  (chain of) addresses (for extra headers), or NULL;
+                          only the first address is used
+  fd                    file descriptor to write the message to
+  sendfn               function for output (transport or verify)
+  wck_flags
+    use_crlf           turn NL into CR LF
+    use_bdat           callback before chunk flush
+  rewrite_rules         chain of header rewriting rules
+  rewrite_existflags    flags for the rewriting rules
+  chunk_cb             transport callback function for data-chunk commands
+
+Returns:                TRUE on success; FALSE on failure.
+*/
+BOOL
+transport_headers_send(int fd, transport_ctx * tctx,
+  BOOL (*sendfn)(int fd, transport_ctx * tctx, uschar * s, int len))
+{
+header_line *h;
+const uschar *list;
+transport_instance * tblock = tctx ? tctx->tblock : NULL;
+address_item * addr = tctx ? tctx->addr : NULL;
+
+/* Then the message's headers. Don't write any that are flagged as "old";
+that means they were rewritten, or are a record of envelope rewriting, or
+were removed (e.g. Bcc). If remove_headers is not null, skip any headers that
+match any entries therein.  It is a colon-sep list; expand the items
+separately and squash any empty ones.
+Then check addr->prop.remove_headers too, provided that addr is not NULL. */
+
+for (h = header_list; h; h = h->next) if (h->type != htype_old)
+  {
+  int i;
+  BOOL include_header = TRUE;
+
+  list = tblock ? tblock->remove_headers : NULL;
+  for (i = 0; i < 2; i++)    /* For remove_headers && addr->prop.remove_headers */
+    {
+    if (list)
+      {
+      int sep = ':';         /* This is specified as a colon-separated list */
+      uschar *s, *ss;
+      while ((s = string_nextinlist(&list, &sep, NULL, 0)))
+       {
+       int len;
+
+       if (i == 0)
+         if (!(s = expand_string(s)) && !expand_string_forcedfail)
+           {
+           errno = ERRNO_CHHEADER_FAIL;
+           return FALSE;
+           }
+       len = s ? Ustrlen(s) : 0;
+       if (strncmpic(h->text, s, len) != 0) continue;
+       ss = h->text + len;
+       while (*ss == ' ' || *ss == '\t') ss++;
+       if (*ss == ':') break;
+       }
+      if (s) { include_header = FALSE; break; }
+      }
+    if (addr) list = addr->prop.remove_headers;
+    }
+
+  /* If this header is to be output, try to rewrite it if there are rewriting
+  rules. */
+
+  if (include_header)
+    {
+    if (tblock && tblock->rewrite_rules)
+      {
+      void *reset_point = store_get(0);
+      header_line *hh;
+
+      if ((hh = rewrite_header(h, NULL, NULL, tblock->rewrite_rules,
+                 tblock->rewrite_existflags, FALSE)))
+       {
+       if (!sendfn(fd, tctx, hh->text, hh->slen)) return FALSE;
+       store_reset(reset_point);
+       continue;     /* With the next header line */
+       }
+      }
+
+    /* Either no rewriting rules, or it didn't get rewritten */
+
+    if (!sendfn(fd, tctx, h->text, h->slen)) return FALSE;
+    }
+
+  /* Header removed */
+
+  else
+    {
+    DEBUG(D_transport) debug_printf("removed header line:\n%s---\n", h->text);
+    }
+  }
+
+/* Add on any address-specific headers. If there are multiple addresses,
+they will all have the same headers in order to be batched. The headers
+are chained in reverse order of adding (so several addresses from the
+same alias might share some of them) but we want to output them in the
+opposite order. This is a bit tedious, but there shouldn't be very many
+of them. We just walk the list twice, reversing the pointers each time,
+but on the second time, write out the items.
+
+Headers added to an address by a router are guaranteed to end with a newline.
+*/
+
+if (addr)
+  {
+  int i;
+  header_line *hprev = addr->prop.extra_headers;
+  header_line *hnext;
+  for (i = 0; i < 2; i++)
+    for (h = hprev, hprev = NULL; h; h = hnext)
+      {
+      hnext = h->next;
+      h->next = hprev;
+      hprev = h;
+      if (i == 1)
+       {
+       if (!sendfn(fd, tctx, h->text, h->slen)) return FALSE;
+       DEBUG(D_transport)
+         debug_printf("added header line(s):\n%s---\n", h->text);
+       }
+      }
+  }
+
+/* If a string containing additional headers exists it is a newline-sep
+list.  Expand each item and write out the result.  This is done last so that
+if it (deliberately or accidentally) isn't in header format, it won't mess
+up any other headers. An empty string or a forced expansion failure are
+noops. An added header string from a transport may not end with a newline;
+add one if it does not. */
+
+if (tblock && (list = CUS tblock->add_headers))
+  {
+  int sep = '\n';
+  uschar * s;
+
+  while ((s = string_nextinlist(&list, &sep, NULL, 0)))
+    if ((s = expand_string(s)))
+      {
+      int len = Ustrlen(s);
+      if (len > 0)
+       {
+       if (!sendfn(fd, tctx, s, len)) return FALSE;
+       if (s[len-1] != '\n' && !sendfn(fd, tctx, US"\n", 1))
+         return FALSE;
+       DEBUG(D_transport)
+         {
+         debug_printf("added header line:\n%s", s);
+         if (s[len-1] != '\n') debug_printf("\n");
+         debug_printf("---\n");
+         }
+       }
+      }
+    else if (!expand_string_forcedfail)
+      { errno = ERRNO_CHHEADER_FAIL; return FALSE; }
+  }
+
+/* Separate headers from body with a blank line */
+
+return sendfn(fd, tctx, US"\n", 1);
+}
+
+
 /*************************************************
 *                Write the message               *
 *************************************************/
@@ -596,30 +806,32 @@ can include timeouts for certain transports, which are requested by setting
 transport_write_timeout non-zero.
 
 Arguments:
-  addr                  (chain of) addresses (for extra headers), or NULL;
-                          only the first address is used
   fd                    file descriptor to write the message to
-  options               bit-wise options:
-    add_return_path       if TRUE, add a "return-path" header
-    add_envelope_to       if TRUE, add a "envelope-to" header
-    add_delivery_date     if TRUE, add a "delivery-date" header
-    use_crlf              if TRUE, turn NL into CR LF
-    end_dot               if TRUE, send a terminating "." line at the end
-    no_headers            if TRUE, omit the headers
-    no_body               if TRUE, omit the body
-  size_limit            if > 0, this is a limit to the size of message written;
-                          it is used when returning messages to their senders,
-                          and is approximate rather than exact, owing to chunk
-                          buffering
-  add_headers           a string containing one or more headers to add; it is
-                          expanded, and must be in correct RFC 822 format as
-                          it is transmitted verbatim; NULL => no additions,
-                          and so does empty string or forced expansion fail
-  remove_headers        a colon-separated list of headers to remove, or NULL
-  check_string          a string to check for at the start of lines, or NULL
-  escape_string         a string to insert in front of any check string
-  rewrite_rules         chain of header rewriting rules
-  rewrite_existflags    flags for the rewriting rules
+  tctx
+    addr                (chain of) addresses (for extra headers), or NULL;
+                          only the first address is used
+    tblock             optional transport instance block (NULL signifies NULL/0):
+      add_headers           a string containing one or more headers to add; it is
+                            expanded, and must be in correct RFC 822 format as
+                            it is transmitted verbatim; NULL => no additions,
+                            and so does empty string or forced expansion fail
+      remove_headers        a colon-separated list of headers to remove, or NULL
+      rewrite_rules         chain of header rewriting rules
+      rewrite_existflags    flags for the rewriting rules
+    options               bit-wise options:
+      add_return_path       if TRUE, add a "return-path" header
+      add_envelope_to       if TRUE, add a "envelope-to" header
+      add_delivery_date     if TRUE, add a "delivery-date" header
+      use_crlf              if TRUE, turn NL into CR LF
+      end_dot               if TRUE, send a terminating "." line at the end
+      no_headers            if TRUE, omit the headers
+      no_body               if TRUE, omit the body
+    check_string          a string to check for at the start of lines, or NULL
+    escape_string         a string to insert in front of any check string
+  size_limit              if > 0, this is a limit to the size of message written;
+                            it is used when returning messages to their senders,
+                            and is approximate rather than exact, owing to chunk
+                            buffering
 
 Returns:                TRUE on success; FALSE (with errno) on failure.
                         In addition, the global variable transport_count
@@ -627,14 +839,9 @@ Returns:                TRUE on success; FALSE (with errno) on failure.
 */
 
 static BOOL
-internal_transport_write_message(address_item *addr, int fd, int options,
-  int size_limit, uschar *add_headers, uschar *remove_headers, uschar *check_string,
-  uschar *escape_string, rewrite_rule *rewrite_rules, int rewrite_existflags)
+internal_transport_write_message(int fd, transport_ctx * tctx, int size_limit)
 {
-int written = 0;
 int len;
-header_line *h;
-BOOL use_crlf  = (options & topt_use_crlf)  != 0;
 
 /* Initialize pointer in output buffer. */
 
@@ -643,39 +850,41 @@ chunk_ptr = deliver_out_buffer;
 /* Set up the data for start-of-line data checking and escaping */
 
 nl_partial_match = -1;
-if (check_string != NULL && escape_string != NULL)
+if (tctx->check_string && tctx->escape_string)
   {
-  nl_check = check_string;
+  nl_check = tctx->check_string;
   nl_check_length = Ustrlen(nl_check);
-  nl_escape = escape_string;
+  nl_escape = tctx->escape_string;
   nl_escape_length = Ustrlen(nl_escape);
   }
-else nl_check_length = nl_escape_length = 0;
+else
+  nl_check_length = nl_escape_length = 0;
 
 /* Whether the escaping mechanism is applied to headers or not is controlled by
 an option (set for SMTP, not otherwise). Negate the length if not wanted till
 after the headers. */
 
-if ((options & topt_escape_headers) == 0) nl_check_length = -nl_check_length;
+if (!(tctx->options & topt_escape_headers))
+  nl_check_length = -nl_check_length;
 
 /* Write the headers if required, including any that have to be added. If there
 are header rewriting rules, apply them. */
 
-if ((options & topt_no_headers) == 0)
+if (!(tctx->options & topt_no_headers))
   {
   /* Add return-path: if requested. */
 
-  if ((options & topt_add_return_path) != 0)
+  if (tctx->options & topt_add_return_path)
     {
     uschar buffer[ADDRESS_MAXLENGTH + 20];
-    sprintf(CS buffer, "Return-path: <%.*s>\n", ADDRESS_MAXLENGTH,
+    int n = sprintf(CS buffer, "Return-path: <%.*s>\n", ADDRESS_MAXLENGTH,
       return_path);
-    if (!write_chunk(fd, buffer, Ustrlen(buffer), use_crlf)) return FALSE;
+    if (!write_chunk(fd, tctx, buffer, n)) return FALSE;
     }
 
   /* Add envelope-to: if requested */
 
-  if ((options & topt_add_envelope_to) != 0)
+  if (tctx->options & topt_add_envelope_to)
     {
     BOOL first = TRUE;
     address_item *p;
@@ -683,225 +892,329 @@ if ((options & topt_no_headers) == 0)
     struct aci *dlist = NULL;
     void *reset_point = store_get(0);
 
-    if (!write_chunk(fd, US"Envelope-to: ", 13, use_crlf)) return FALSE;
+    if (!write_chunk(fd, tctx, US"Envelope-to: ", 13)) return FALSE;
 
     /* Pick up from all the addresses. The plist and dlist variables are
     anchors for lists of addresses already handled; they have to be defined at
     this level becuase write_env_to() calls itself recursively. */
 
-    for (p = addr; p != NULL; p = p->next)
-      {
-      if (!write_env_to(p, &plist, &dlist, &first, fd, use_crlf)) return FALSE;
-      }
+    for (p = tctx->addr; p; p = p->next)
+      if (!write_env_to(p, &plist, &dlist, &first, fd, tctx))
+       return FALSE;
 
     /* Add a final newline and reset the store used for tracking duplicates */
 
-    if (!write_chunk(fd, US"\n", 1, use_crlf)) return FALSE;
+    if (!write_chunk(fd, tctx, US"\n", 1)) return FALSE;
     store_reset(reset_point);
     }
 
   /* Add delivery-date: if requested. */
 
-  if ((options & topt_add_delivery_date) != 0)
+  if (tctx->options & topt_add_delivery_date)
     {
     uschar buffer[100];
-    sprintf(CS buffer, "Delivery-date: %s\n", tod_stamp(tod_full));
-    if (!write_chunk(fd, buffer, Ustrlen(buffer), use_crlf)) return FALSE;
+    int n = sprintf(CS buffer, "Delivery-date: %s\n", tod_stamp(tod_full));
+    if (!write_chunk(fd, tctx, buffer, n)) return FALSE;
     }
 
   /* Then the message's headers. Don't write any that are flagged as "old";
   that means they were rewritten, or are a record of envelope rewriting, or
   were removed (e.g. Bcc). If remove_headers is not null, skip any headers that
-  match any entries therein. Then check addr->p.remove_headers too, provided that
+  match any entries therein. Then check addr->prop.remove_headers too, provided that
   addr is not NULL. */
 
-  if (remove_headers != NULL)
+  if (!transport_headers_send(fd, tctx, &write_chunk))
+    return FALSE;
+  }
+
+/* When doing RFC3030 CHUNKING output, work out how much data will be in the
+last BDAT, consisting of the current write_chunk() output buffer fill
+(optimally, all of the headers - but it does not matter if we already had to
+flush that buffer with non-last BDAT prependix) plus the amount of body data
+(as expanded for CRLF lines).  Then create and write the BDAT, and ensure
+that further use of write_chunk() will not prepend BDATs.
+The first BDAT written will also first flush any outstanding MAIL and RCPT
+commands which were buffered thans to PIPELINING.
+Commands go out (using a send()) from a different buffer to data (using a
+write()).  They might not end up in the same TCP segment, which is
+suboptimal. */
+
+if (tctx->options & topt_use_bdat)
+  {
+  off_t fsize;
+  int hsize, size;
+
+  if ((hsize = chunk_ptr - deliver_out_buffer) < 0)
+    hsize = 0;
+  if (!(tctx->options & topt_no_body))
     {
-    uschar *s = expand_string(remove_headers);
-    if (s == NULL && !expand_string_forcedfail)
-      {
-      errno = ERRNO_CHHEADER_FAIL;
+    if ((fsize = lseek(deliver_datafile, 0, SEEK_END)) < 0) return FALSE;
+    fsize -= SPOOL_DATA_START_OFFSET;
+    if (size_limit > 0  &&  fsize > size_limit)
+      fsize = size_limit;
+    size = hsize + fsize;
+    if (tctx->options & topt_use_crlf)
+      size += body_linecount;  /* account for CRLF-expansion */
+    }
+
+  /* If the message is large, emit first a non-LAST chunk with just the
+  headers, and reap the command responses.  This lets us error out early
+  on RCPT rejects rather than sending megabytes of data.  Include headers
+  on the assumption they are cheap enough and some clever implementations
+  might errorcheck them too, on-the-fly, and reject that chunk. */
+
+  if (size > DELIVER_OUT_BUFFER_SIZE && hsize > 0)
+    {
+    if (  tctx->chunk_cb(fd, tctx, hsize, 0) != OK
+       || !transport_write_block(fd, deliver_out_buffer, hsize)
+       || tctx->chunk_cb(fd, tctx, 0, tc_reap_prev) != OK
+       )
       return FALSE;
-      }
-    remove_headers = s;
+    chunk_ptr = deliver_out_buffer;
+    size -= hsize;
     }
 
-  for (h = header_list; h != NULL; h = h->next)
+  /* Emit a LAST datachunk command. */
+
+  if (tctx->chunk_cb(fd, tctx, size, tc_chunk_last) != OK)
+    return FALSE;
+
+  tctx->options &= ~topt_use_bdat;
+  }
+
+/* If the body is required, ensure that the data for check strings (formerly
+the "from hack") is enabled by negating the length if necessary. (It will be
+negative in cases where it isn't to apply to the headers). Then ensure the body
+is positioned at the start of its file (following the message id), then write
+it, applying the size limit if required. */
+
+if (!(tctx->options & topt_no_body))
+  {
+  int size = size_limit;
+
+  nl_check_length = abs(nl_check_length);
+  nl_partial_match = 0;
+  if (lseek(deliver_datafile, SPOOL_DATA_START_OFFSET, SEEK_SET) < 0)
+    return FALSE;
+  while (  (len = MAX(DELIVER_IN_BUFFER_SIZE, size)) > 0
+       && (len = read(deliver_datafile, deliver_in_buffer, len)) > 0)
     {
-    int i;
-    uschar *list = NULL;
-    BOOL include_header;
+    if (!write_chunk(fd, tctx, deliver_in_buffer, len))
+      return FALSE;
+    size -= len;
+    }
 
-    if (h->type == htype_old) continue;
+  /* A read error on the body will have left len == -1 and errno set. */
 
-    include_header = TRUE;
-    list = remove_headers;
+  if (len != 0) return FALSE;
+  }
 
-    for (i = 0; i < 2; i++)    /* For remove_headers && addr->p.remove_headers */
-      {
-      if (list != NULL)
-        {
-        int sep = ':';         /* This is specified as a colon-separated list */
-        uschar *s, *ss;
-        uschar buffer[128];
-        while ((s = string_nextinlist(&list, &sep, buffer, sizeof(buffer)))
-                != NULL)
-          {
-          int len = Ustrlen(s);
-          if (strncmpic(h->text, s, len) != 0) continue;
-          ss = h->text + len;
-          while (*ss == ' ' || *ss == '\t') ss++;
-          if (*ss == ':') break;
-          }
-        if (s != NULL) { include_header = FALSE; break; }
-        }
-      if (addr != NULL) list = addr->p.remove_headers;
-      }
+/* Finished with the check string */
 
-    /* If this header is to be output, try to rewrite it if there are rewriting
-    rules. */
+nl_check_length = nl_escape_length = 0;
 
-    if (include_header)
-      {
-      if (rewrite_rules != NULL)
-        {
-        void *reset_point = store_get(0);
-        header_line *hh =
-          rewrite_header(h, NULL, NULL, rewrite_rules, rewrite_existflags,
-            FALSE);
-        if (hh != NULL)
-          {
-          if (!write_chunk(fd, hh->text, hh->slen, use_crlf)) return FALSE;
-          store_reset(reset_point);
-          continue;     /* With the next header line */
-          }
-        }
+/* If requested, add a terminating "." line (SMTP output). */
 
-      /* Either no rewriting rules, or it didn't get rewritten */
+if (tctx->options & topt_end_dot && !write_chunk(fd, tctx, US".\n", 2))
+  return FALSE;
 
-      if (!write_chunk(fd, h->text, h->slen, use_crlf)) return FALSE;
-      }
+/* Write out any remaining data in the buffer before returning. */
 
-    /* Header removed */
+return (len = chunk_ptr - deliver_out_buffer) <= 0 ||
+  transport_write_block(fd, deliver_out_buffer, len);
+}
 
-    else
-      {
-      DEBUG(D_transport) debug_printf("removed header line:\n%s---\n",
-        h->text);
-      }
-    }
 
-  /* Add on any address-specific headers. If there are multiple addresses,
-  they will all have the same headers in order to be batched. The headers
-  are chained in reverse order of adding (so several addresses from the
-  same alias might share some of them) but we want to output them in the
-  opposite order. This is a bit tedious, but there shouldn't be very many
-  of them. We just walk the list twice, reversing the pointers each time,
-  but on the second time, write out the items. */
+#ifndef DISABLE_DKIM
 
-  if (addr != NULL)
-    {
-    int i;
-    header_line *hprev = addr->p.extra_headers;
-    header_line *hnext;
-    for (i = 0; i < 2; i++)
-      {
-      for (h = hprev, hprev = NULL; h != NULL; h = hnext)
-        {
-        hnext = h->next;
-        h->next = hprev;
-        hprev = h;
-        if (i == 1)
-          {
-          if (!write_chunk(fd, h->text, h->slen, use_crlf)) return FALSE;
-          DEBUG(D_transport)
-            debug_printf("added header line(s):\n%s---\n", h->text);
-          }
-        }
-      }
-    }
+/***************************************************************************************************
+*    External interface to write the message, while signing it with DKIM and/or Domainkeys         *
+***************************************************************************************************/
 
-  /* If a string containing additional headers exists, expand it and write
-  out the result. This is done last so that if it (deliberately or accidentally)
-  isn't in header format, it won't mess up any other headers. An empty string
-  or a forced expansion failure are noops. */
+/* This function is a wrapper around transport_write_message().
+   It is only called from the smtp transport if DKIM or Domainkeys support
+   is compiled in.  The function sets up a replacement fd into a -K file,
+   then calls the normal function. This way, the exact bits that exim would
+   have put "on the wire" will end up in the file (except for TLS
+   encapsulation, which is the very very last thing). When we are done
+   signing the file, send the signed message down the original fd (or TLS fd).
 
-  if (add_headers != NULL)
-    {
-    uschar *s = expand_string(add_headers);
-    if (s == NULL)
-      {
-      if (!expand_string_forcedfail)
-        {
-        errno = ERRNO_CHHEADER_FAIL;
-        return FALSE;
-        }
-      }
-    else
-      {
-      int len = Ustrlen(s);
-      if (len > 0)
-        {
-        if (!write_chunk(fd, s, len, use_crlf)) return FALSE;
-        if (s[len-1] != '\n' && !write_chunk(fd, US"\n", 1, use_crlf))
-          return FALSE;
-        DEBUG(D_transport)
-          debug_printf("added header line(s):\n%s---\n", s);
-        }
-      }
-    }
+Arguments:
+  as for internal_transport_write_message() above, with additional arguments
+  for DKIM.
 
-  /* Separate headers from body with a blank line */
+Returns:       TRUE on success; FALSE (with errno) for any failure
+*/
 
-  if (!write_chunk(fd, US"\n", 1, use_crlf)) return FALSE;
+BOOL
+dkim_transport_write_message(int out_fd, transport_ctx * tctx,
+  struct ob_dkim * dkim)
+{
+int dkim_fd;
+int save_errno = 0;
+BOOL rc;
+uschar * dkim_spool_name;
+int sread = 0;
+int wwritten = 0;
+uschar *dkim_signature = NULL;
+int siglen = 0;
+off_t k_file_size;
+int options;
+
+/* If we can't sign, just call the original function. */
+
+if (!(dkim->dkim_private_key && dkim->dkim_domain && dkim->dkim_selector))
+  return transport_write_message(out_fd, tctx, 0);
+
+dkim_spool_name = spool_fname(US"input", message_subdir, message_id,
+                   string_sprintf("-%d-K", (int)getpid()));
+
+if ((dkim_fd = Uopen(dkim_spool_name, O_RDWR|O_CREAT|O_TRUNC, SPOOL_MODE)) < 0)
+  {
+  /* Can't create spool file. Ugh. */
+  rc = FALSE;
+  save_errno = errno;
+  goto CLEANUP;
   }
 
-/* If the body is required, ensure that the data for check strings (formerly
-the "from hack") is enabled by negating the length if necessary. (It will be
-negative in cases where it isn't to apply to the headers). Then ensure the body
-is positioned at the start of its file (following the message id), then write
-it, applying the size limit if required. */
+/* Call original function to write the -K file; does the CRLF expansion */
 
-if ((options & topt_no_body) == 0)
+options = tctx->options;
+tctx->options &= ~topt_use_bdat;
+rc = transport_write_message(dkim_fd, tctx, 0);
+tctx->options = options;
+
+/* Save error state. We must clean up before returning. */
+if (!rc)
   {
-  nl_check_length = abs(nl_check_length);
-  nl_partial_match = 0;
-  lseek(deliver_datafile, SPOOL_DATA_START_OFFSET, SEEK_SET);
-  while ((len = read(deliver_datafile, deliver_in_buffer,
-           DELIVER_IN_BUFFER_SIZE)) > 0)
-    {
-    if (!write_chunk(fd, deliver_in_buffer, len, use_crlf)) return FALSE;
-    if (size_limit > 0)
+  save_errno = errno;
+  goto CLEANUP;
+  }
+
+/* Rewind file and feed it to the goats^W DKIM lib */
+lseek(dkim_fd, 0, SEEK_SET);
+dkim_signature = dkim_exim_sign(dkim_fd,
+                               dkim->dkim_private_key,
+                               dkim->dkim_domain,
+                               dkim->dkim_selector,
+                               dkim->dkim_canon,
+                               dkim->dkim_sign_headers);
+if (dkim_signature)
+  siglen = Ustrlen(dkim_signature);
+else if (dkim->dkim_strict)
+  {
+  uschar *dkim_strict_result = expand_string(dkim->dkim_strict);
+  if (dkim_strict_result)
+    if ( (strcmpic(dkim->dkim_strict,US"1") == 0) ||
+        (strcmpic(dkim->dkim_strict,US"true") == 0) )
       {
-      written += len;
-      if (written > size_limit)
-        {
-        len = 0;    /* Pretend EOF */
-        break;
-        }
+      /* Set errno to something halfway meaningful */
+      save_errno = EACCES;
+      log_write(0, LOG_MAIN, "DKIM: message could not be signed,"
+       " and dkim_strict is set. Deferring message delivery.");
+      rc = FALSE;
+      goto CLEANUP;
       }
+  }
+
+#ifndef HAVE_LINUX_SENDFILE
+if (options & topt_use_bdat)
+#endif
+  k_file_size = lseek(dkim_fd, 0, SEEK_END); /* Fetch file size */
+
+if (options & topt_use_bdat)
+  {
+
+  /* On big messages output a precursor chunk to get any pipelined
+  MAIL & RCPT commands flushed, then reap the responses so we can
+  error out on RCPT rejects before sending megabytes. */
+
+  if (siglen + k_file_size > DELIVER_OUT_BUFFER_SIZE && siglen > 0)
+    {
+    if (  tctx->chunk_cb(out_fd, tctx, siglen, 0) != OK
+       || !transport_write_block(out_fd, dkim_signature, siglen)
+       || tctx->chunk_cb(out_fd, tctx, 0, tc_reap_prev) != OK
+       )
+      goto err;
+    siglen = 0;
     }
 
-  /* Finished with the check string */
+  if (tctx->chunk_cb(out_fd, tctx, siglen + k_file_size, tc_chunk_last) != OK)
+    goto err;
+  }
 
-  nl_check_length = nl_escape_length = 0;
+if(siglen > 0 && !transport_write_block(out_fd, dkim_signature, siglen))
+  goto err;
 
-  /* A read error on the body will have left len == -1 and errno set. */
+#ifdef HAVE_LINUX_SENDFILE
+/* We can use sendfile() to shove the file contents
+   to the socket. However only if we don't use TLS,
+   as then there's another layer of indirection
+   before the data finally hits the socket. */
+if (tls_out.active != out_fd)
+  {
+  ssize_t copied = 0;
+  off_t offset = 0;
 
-  if (len != 0) return FALSE;
+  /* Rewind file */
+  lseek(dkim_fd, 0, SEEK_SET);
 
-  /* If requested, add a terminating "." line (SMTP output). */
+  while(copied >= 0 && offset < k_file_size)
+    copied = sendfile(out_fd, dkim_fd, &offset, k_file_size - offset);
+  if (copied < 0)
+    goto err;
+  }
+else
 
-  if ((options & topt_end_dot) != 0 && !write_chunk(fd, US".\n", 2, use_crlf))
-    return FALSE;
+#endif
+
+  {
+  /* Rewind file */
+  lseek(dkim_fd, 0, SEEK_SET);
+
+  /* Send file down the original fd */
+  while((sread = read(dkim_fd, deliver_out_buffer, DELIVER_OUT_BUFFER_SIZE)) >0)
+    {
+    char *p = deliver_out_buffer;
+    /* write the chunk */
+
+    while (sread)
+      {
+#ifdef SUPPORT_TLS
+      wwritten = tls_out.active == out_fd
+       ? tls_write(FALSE, US p, sread)
+       : write(out_fd, p, sread);
+#else
+      wwritten = write(out_fd, p, sread);
+#endif
+      if (wwritten == -1)
+       goto err;
+      p += wwritten;
+      sread -= wwritten;
+      }
+    }
+
+  if (sread == -1)
+    {
+    save_errno = errno;
+    rc = FALSE;
+    }
   }
 
-/* Write out any remaining data in the buffer before returning. */
+CLEANUP:
+  /* unlink -K file */
+  (void)close(dkim_fd);
+  Uunlink(dkim_spool_name);
+  errno = save_errno;
+  return rc;
 
-return (len = chunk_ptr - deliver_out_buffer) <= 0 ||
-  transport_write_block(fd, deliver_out_buffer, len);
+err:
+  save_errno = errno;
+  rc = FALSE;
+  goto CLEANUP;
 }
 
+#endif
 
 
 
@@ -922,37 +1235,40 @@ Returns:       TRUE on success; FALSE (with errno) for any failure
 */
 
 BOOL
-transport_write_message(address_item *addr, int fd, int options,
-  int size_limit, uschar *add_headers, uschar *remove_headers,
-  uschar *check_string, uschar *escape_string, rewrite_rule *rewrite_rules,
-  int rewrite_existflags)
+transport_write_message(int fd, transport_ctx * tctx, int size_limit)
 {
-BOOL use_crlf;
+unsigned wck_flags;
 BOOL last_filter_was_NL = TRUE;
 int rc, len, yield, fd_read, fd_write, save_errno;
-int pfd[2];
+int pfd[2] = {-1, -1};
 pid_t filter_pid, write_pid;
+static transport_ctx dummy_tctx = {0};
+
+if (!tctx) tctx = &dummy_tctx;
+
+transport_filter_timed_out = FALSE;
 
 /* If there is no filter command set up, call the internal function that does
 the actual work, passing it the incoming fd, and return its result. */
 
-if (transport_filter_argv == NULL)
-  return internal_transport_write_message(addr, fd, options, size_limit,
-    add_headers, remove_headers, check_string, escape_string,
-    rewrite_rules, rewrite_existflags);
+if (  !transport_filter_argv
+   || !*transport_filter_argv
+   || !**transport_filter_argv
+   )
+  return internal_transport_write_message(fd, tctx, size_limit);
 
 /* Otherwise the message must be written to a filter process and read back
 before being written to the incoming fd. First set up the special processing to
 be done during the copying. */
 
-use_crlf  = (options & topt_use_crlf) != 0;
+wck_flags = tctx->options & topt_use_crlf;
 nl_partial_match = -1;
 
-if (check_string != NULL && escape_string != NULL)
+if (tctx->check_string && tctx->escape_string)
   {
-  nl_check = check_string;
+  nl_check = tctx->check_string;
   nl_check_length = Ustrlen(nl_check);
-  nl_escape = escape_string;
+  nl_escape = tctx->escape_string;
   nl_escape_length = Ustrlen(nl_escape);
   }
 else nl_check_length = nl_escape_length = 0;
@@ -969,14 +1285,14 @@ save_errno = 0;
 yield = FALSE;
 write_pid = (pid_t)(-1);
 
-fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) | FD_CLOEXEC);
-filter_pid = child_open(transport_filter_argv, NULL, 077, &fd_write, &fd_read,
-  FALSE);
-fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) & ~FD_CLOEXEC);
+(void)fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) | FD_CLOEXEC);
+filter_pid = child_open(USS transport_filter_argv, NULL, 077,
&fd_write, &fd_read, FALSE);
+(void)fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) & ~FD_CLOEXEC);
 if (filter_pid < 0) goto TIDY_UP;      /* errno set */
 
 DEBUG(D_transport)
-  debug_printf("process %d running as transport filter: write=%d read=%d\n",
+  debug_printf("process %d running as transport filter: fd_write=%d fd_read=%d\n",
     (int)filter_pid, fd_write, fd_read);
 
 /* Fork subprocess to write the message to the filter, and return the result
@@ -987,25 +1303,32 @@ if (pipe(pfd) != 0) goto TIDY_UP;      /* errno set */
 if ((write_pid = fork()) == 0)
   {
   BOOL rc;
-  close(fd_read);
-  close(pfd[pipe_read]);
+  (void)close(fd_read);
+  (void)close(pfd[pipe_read]);
   nl_check_length = nl_escape_length = 0;
-  rc = internal_transport_write_message(addr, fd_write,
-    (options & ~(topt_use_crlf | topt_end_dot)),
-    size_limit, add_headers, remove_headers, NULL, NULL,
-    rewrite_rules, rewrite_existflags);
+
+  tctx->check_string = tctx->escape_string = NULL;
+  tctx->options &= ~(topt_use_crlf | topt_end_dot | topt_use_bdat);
+
+  rc = internal_transport_write_message(fd_write, tctx, size_limit);
+
   save_errno = errno;
-  write(pfd[pipe_write], (void *)&rc, sizeof(BOOL));
-  write(pfd[pipe_write], (void *)&save_errno, sizeof(int));
-  write(pfd[pipe_write], (void *)&(addr->more_errno), sizeof(int));
+  if (  write(pfd[pipe_write], (void *)&rc, sizeof(BOOL))
+        != sizeof(BOOL)
+     || write(pfd[pipe_write], (void *)&save_errno, sizeof(int))
+        != sizeof(int)
+     || write(pfd[pipe_write], (void *)&tctx->addr->more_errno, sizeof(int))
+        != sizeof(int)
+     )
+    rc = FALSE;        /* compiler quietening */
   _exit(0);
   }
 save_errno = errno;
 
 /* Parent process: close our copy of the writing subprocess' pipes. */
 
-close(pfd[pipe_write]);
-close(fd_write);
+(void)close(pfd[pipe_write]);
+(void)close(fd_write);
 fd_write = -1;
 
 /* Writing process creation failed */
@@ -1045,6 +1368,7 @@ for (;;)
   if (sigalrm_seen)
     {
     errno = ETIMEDOUT;
+    transport_filter_timed_out = TRUE;
     goto TIDY_UP;
     }
 
@@ -1053,7 +1377,7 @@ for (;;)
 
   if (len > 0)
     {
-    if (!write_chunk(fd, deliver_in_buffer, len, use_crlf)) goto TIDY_UP;
+    if (!write_chunk(fd, tctx, deliver_in_buffer, len)) goto TIDY_UP;
     last_filter_was_NL = (deliver_in_buffer[len-1] == '\n');
     }
 
@@ -1074,8 +1398,8 @@ sure. Also apply a paranoia timeout. */
 TIDY_UP:
 save_errno = errno;
 
-close(fd_read);
-if (fd_write > 0) close(fd_write);
+(void)close(fd_read);
+if (fd_write > 0) (void)close(fd_write);
 
 if (!yield)
   {
@@ -1090,12 +1414,12 @@ if (filter_pid > 0 && (rc = child_close(filter_pid, 30)) != 0 && yield)
   {
   yield = FALSE;
   save_errno = ERRNO_FILTER_FAIL;
-  addr->more_errno = rc;
+  tctx->addr->more_errno = rc;
   DEBUG(D_transport) debug_printf("filter process returned %d\n", rc);
   }
 
 /* Wait for the writing process to complete. If it ends successfully,
-read the results from its pipe, provided we haven't already had a filter 
+read the results from its pipe, provided we haven't already had a filter
 process failure. */
 
 DEBUG(D_transport) debug_printf("waiting for writing process\n");
@@ -1104,14 +1428,14 @@ if (write_pid > 0)
   rc = child_close(write_pid, 30);
   if (yield)
     {
-    if (rc == 0)   
+    if (rc == 0)
       {
       BOOL ok;
-      read(pfd[pipe_read], (void *)&ok, sizeof(BOOL));
+      int dummy = read(pfd[pipe_read], (void *)&ok, sizeof(BOOL));
       if (!ok)
         {
-        read(pfd[pipe_read], (void *)&save_errno, sizeof(int));
-        read(pfd[pipe_read], (void *)&(addr->more_errno), sizeof(int));
+        dummy = read(pfd[pipe_read], (void *)&save_errno, sizeof(int));
+        dummy = read(pfd[pipe_read], (void *)&(tctx->addr->more_errno), sizeof(int));
         yield = FALSE;
         }
       }
@@ -1119,12 +1443,12 @@ if (write_pid > 0)
       {
       yield = FALSE;
       save_errno = ERRNO_FILTER_FAIL;
-      addr->more_errno = rc;
+      tctx->addr->more_errno = rc;
       DEBUG(D_transport) debug_printf("writing process returned %d\n", rc);
       }
-    }   
+    }
   }
-close(pfd[pipe_read]);
+(void)close(pfd[pipe_read]);
 
 /* If there have been no problems we can now add the terminating "." if this is
 SMTP output, turning off escaping beforehand. If the last character from the
@@ -1133,28 +1457,27 @@ filter was not NL, insert a NL to make the SMTP protocol work. */
 if (yield)
   {
   nl_check_length = nl_escape_length = 0;
-  if ((options & topt_end_dot) != 0 && (last_filter_was_NL?
-        !write_chunk(fd, US".\n", 2, use_crlf) :
-        !write_chunk(fd, US"\n.\n", 3, use_crlf)))
-    {
+  if (  tctx->options & topt_end_dot
+     && ( last_filter_was_NL
+        ? !write_chunk(fd, tctx, US".\n", 2)
+       : !write_chunk(fd, tctx, US"\n.\n", 3)
+     )  )
     yield = FALSE;
-    }
 
   /* Write out any remaining data in the buffer. */
 
   else
-    {
-    yield = (len = chunk_ptr - deliver_out_buffer) <= 0 ||
-      transport_write_block(fd, deliver_out_buffer, len);
-    }
+    yield = (len = chunk_ptr - deliver_out_buffer) <= 0
+         || transport_write_block(fd, deliver_out_buffer, len);
   }
-else errno = save_errno;      /* From some earlier error */
+else
+  errno = save_errno;      /* From some earlier error */
 
 DEBUG(D_transport)
   {
   debug_printf("end of filtering transport writing: yield=%d\n", yield);
   if (!yield)
-    debug_printf("errno=%d more_errno=%d\n", errno, addr->more_errno);
+    debug_printf("errno=%d more_errno=%d\n", errno, tctx->addr->more_errno);
   }
 
 return yield;
@@ -1193,8 +1516,7 @@ better.
 Old records should eventually get swept up by the exim_tidydb utility.
 
 Arguments:
-  hostlist  list of hosts that this message could be sent to;
-              the update_waiting flag is set if a host is to be noted
+  hostlist  list of hosts that this message could be sent to
   tpname    name of the transport
 
 Returns:    nothing
@@ -1204,11 +1526,13 @@ void
 transport_update_waiting(host_item *hostlist, uschar *tpname)
 {
 uschar buffer[256];
-uschar *prevname = US"";
+const uschar *prevname = US"";
 host_item *host;
 open_db dbblock;
 open_db *dbm_file;
 
+DEBUG(D_transport) debug_printf("updating wait-%s database\n", tpname);
+
 /* Open the database for this transport */
 
 sprintf(CS buffer, "wait-%.200s", tpname);
@@ -1216,8 +1540,7 @@ dbm_file = dbfn_open(buffer, O_RDWR, &dbblock, TRUE);
 if (dbm_file == NULL) return;
 
 /* Scan the list of hosts for which this message is waiting, and ensure
-that the message id is in each host record for those that have the
-update_waiting flag set. */
+that the message id is in each host record. */
 
 for (host = hostlist; host!= NULL; host = host->next)
   {
@@ -1226,10 +1549,6 @@ for (host = hostlist; host!= NULL; host = host->next)
   uschar *s;
   int i, host_length;
 
-  /* Skip if the update_waiting flag is not set. */
-
-  if (!host->update_waiting) continue;
-
   /* Skip if this is the same host as we just processed; otherwise remember
   the name for next time. */
 
@@ -1279,7 +1598,11 @@ for (host = hostlist; host!= NULL; host = host->next)
 
   /* If this message is already in a record, no need to update. */
 
-  if (already) continue;
+  if (already)
+    {
+    DEBUG(D_transport) debug_printf("already listed for %s\n", host->name);
+    continue;
+    }
 
 
   /* If this record is full, write it out with a new name constructed
@@ -1315,6 +1638,7 @@ for (host = hostlist; host!= NULL; host = host->next)
   /* Update the database */
 
   dbfn_write(dbm_file, host->name, host_record, sizeof(dbdata_wait) + host_length);
+  DEBUG(D_transport) debug_printf("added to list for %s\n", host->name);
   }
 
 /* All now done */
@@ -1343,20 +1667,32 @@ Arguments:
                        as set by the caller transport
   new_message_id     set to the message id of a waiting message
   more               set TRUE if there are yet more messages waiting
+  oicf_func          function to call to validate if it is ok to send
+                     to this message_id from the current instance.
+  oicf_data          opaque data for oicf_func
 
 Returns:             TRUE if new_message_id set; FALSE otherwise
 */
 
+typedef struct msgq_s
+{
+    uschar  message_id [MESSAGE_ID_LENGTH + 1];
+    BOOL    bKeep;
+} msgq_t;
+
 BOOL
-transport_check_waiting(uschar *transport_name, uschar *hostname,
-  int local_message_max, uschar *new_message_id, BOOL *more)
+transport_check_waiting(const uschar *transport_name, const uschar *hostname,
+  int local_message_max, uschar *new_message_id, BOOL *more, oicf oicf_func, void *oicf_data)
 {
 dbdata_wait *host_record;
-int host_length, path_len;
+int host_length;
 open_db dbblock;
 open_db *dbm_file;
 uschar buffer[256];
 
+int         i;
+struct stat statbuf;
+
 *more = FALSE;
 
 DEBUG(D_transport)
@@ -1385,8 +1721,7 @@ if (dbm_file == NULL) return FALSE;
 
 /* See if there is a record for this host; if not, there's nothing to do. */
 
-host_record = dbfn_read(dbm_file, hostname);
-if (host_record == NULL)
+if (!(host_record = dbfn_read(dbm_file, hostname)))
   {
   dbfn_close(dbm_file);
   DEBUG(D_transport) debug_printf("no messages waiting for %s\n", hostname);
@@ -1409,58 +1744,106 @@ until one is found for which a spool file actually exists. If the record gets
 emptied, delete it and continue with any continuation records that may exist.
 */
 
-host_length = host_record->count * MESSAGE_ID_LENGTH;
+/* For Bug 1141, I refactored this major portion of the routine, it is risky
+but the 1 off will remain without it.  This code now allows me to SKIP over
+a message I do not want to send out on this run.  */
 
-/* Loop to handle continuation host records in the database */
+host_length = host_record->count * MESSAGE_ID_LENGTH;
 
-for (;;)
+while (1)
   {
-  BOOL found = FALSE;
+  msgq_t      *msgq;
+  int         msgq_count = 0;
+  int         msgq_actual = 0;
+  BOOL        bFound = FALSE;
+  BOOL        bContinuation = FALSE;
+
+  /* create an array to read entire message queue into memory for processing  */
 
-  sprintf(CS buffer, "%s/input/", spool_directory);
-  path_len = Ustrlen(buffer);
+  msgq = (msgq_t*) malloc(sizeof(msgq_t) * host_record->count);
+  msgq_count = host_record->count;
+  msgq_actual = msgq_count;
 
-  for (host_length -= MESSAGE_ID_LENGTH; host_length >= 0;
-       host_length -= MESSAGE_ID_LENGTH)
+  for (i = 0; i < host_record->count; ++i)
     {
-    struct stat statbuf;
-    Ustrncpy(new_message_id, host_record->text + host_length,
+    msgq[i].bKeep = TRUE;
+
+    Ustrncpy(msgq[i].message_id, host_record->text + (i * MESSAGE_ID_LENGTH),
       MESSAGE_ID_LENGTH);
-    new_message_id[MESSAGE_ID_LENGTH] = 0;
+    msgq[i].message_id[MESSAGE_ID_LENGTH] = 0;
+    }
 
-    if (split_spool_directory)
-      sprintf(CS(buffer + path_len), "%c/%s-D", new_message_id[5], new_message_id);
-    else
-      sprintf(CS(buffer + path_len), "%s-D", new_message_id);
+  /* first thing remove current message id if it exists */
+
+  for (i = 0; i < msgq_count; ++i)
+    if (Ustrcmp(msgq[i].message_id, message_id) == 0)
+      {
+      msgq[i].bKeep = FALSE;
+      break;
+      }
 
-    /* The listed message may be the one we are currently processing. If
-    so, we want to remove it from the list without doing anything else.
-    If not, do a stat to see if it is an existing message. If it is, break
-    the loop to handle it. No need to bother about locks; as this is all
-    "hint" processing, it won't matter if it doesn't exist by the time exim
-    actually tries to deliver it. */
+  /* now find the next acceptable message_id */
 
-    if (Ustrcmp(new_message_id, message_id) != 0 &&
-        Ustat(buffer, &statbuf) == 0)
+  for (i = msgq_count - 1; i >= 0; --i) if (msgq[i].bKeep)
+    {
+    uschar subdir[2];
+
+    subdir[0] = split_spool_directory ? msgq[i].message_id[5] : 0;
+    subdir[1] = 0;
+
+    if (Ustat(spool_fname(US"input", subdir, msgq[i].message_id, US"-D"),
+             &statbuf) != 0)
+      msgq[i].bKeep = FALSE;
+    else if (!oicf_func || oicf_func(msgq[i].message_id, oicf_data))
       {
-      found = TRUE;
+      Ustrcpy(new_message_id, msgq[i].message_id);
+      msgq[i].bKeep = FALSE;
+      bFound = TRUE;
       break;
       }
     }
 
-  /* If we have removed all the message ids from the record delete the record.
-  If there is a continuation record, fetch it and remove it from the file,
-  as it will be rewritten as the main record. Repeat in the case of an
-  empty continuation. */
+  /* re-count */
+  for (msgq_actual = 0, i = 0; i < msgq_count; ++i)
+    if (msgq[i].bKeep)
+      msgq_actual++;
+
+  /* reassemble the host record, based on removed message ids, from in
+  memory queue  */
+
+  if (msgq_actual <= 0)
+    {
+    host_length = 0;
+    host_record->count = 0;
+    }
+  else
+    {
+    host_length = msgq_actual * MESSAGE_ID_LENGTH;
+    host_record->count = msgq_actual;
+
+    if (msgq_actual < msgq_count)
+      {
+      int new_count;
+      for (new_count = 0, i = 0; i < msgq_count; ++i)
+       if (msgq[i].bKeep)
+         Ustrncpy(&host_record->text[new_count++ * MESSAGE_ID_LENGTH],
+           msgq[i].message_id, MESSAGE_ID_LENGTH);
+
+      host_record->text[new_count * MESSAGE_ID_LENGTH] = 0;
+      }
+    }
+
+/* Jeremy: check for a continuation record, this code I do not know how to
+test but the code should work */
 
   while (host_length <= 0)
     {
     int i;
-    dbdata_wait *newr = NULL;
+    dbdata_wait * newr = NULL;
 
     /* Search for a continuation */
 
-    for (i = host_record->sequence - 1; i >= 0 && newr == NULL; i--)
+    for (i = host_record->sequence - 1; i >= 0 && !newr; i--)
       {
       sprintf(CS buffer, "%.200s:%d", hostname, i);
       newr = dbfn_read(dbm_file, buffer);
@@ -1468,7 +1851,7 @@ for (;;)
 
     /* If no continuation, delete the current and break the loop */
 
-    if (newr == NULL)
+    if (!newr)
       {
       dbfn_delete(dbm_file, hostname);
       break;
@@ -1479,11 +1862,15 @@ for (;;)
     dbfn_delete(dbm_file, buffer);
     host_record = newr;
     host_length = host_record->count * MESSAGE_ID_LENGTH;
-    }
 
-  /* If we found an existing message, break the continuation loop. */
+    bContinuation = TRUE;
+    }
 
-  if (found) break;
+  if (bFound)          /* Usual exit from main loop */
+    {
+    free (msgq);
+    break;
+    }
 
   /* If host_length <= 0 we have emptied a record and not found a good message,
   and there are no continuation records. Otherwise there is a continuation
@@ -1495,7 +1882,20 @@ for (;;)
     DEBUG(D_transport) debug_printf("waiting messages already delivered\n");
     return FALSE;
     }
-  }
+
+  /* we were not able to find an acceptable message, nor was there a
+   * continuation record.  So bug out, outer logic will clean this up.
+   */
+
+  if (!bContinuation)
+    {
+    Ustrcpy(new_message_id, message_id);
+    dbfn_close(dbm_file);
+    return FALSE;
+    }
+
+  free(msgq);
+  }            /* we need to process a continuation record */
 
 /* Control gets here when an existing message has been encountered; its
 id is in new_message_id, and host_length is the revised length of the
@@ -1505,6 +1905,7 @@ record if required, close the database, and return TRUE. */
 if (host_length > 0)
   {
   host_record->count = host_length/MESSAGE_ID_LENGTH;
+
   dbfn_write(dbm_file, hostname, host_record, (int)sizeof(dbdata_wait) + host_length);
   *more = TRUE;
   }
@@ -1513,8 +1914,6 @@ dbfn_close(dbm_file);
 return TRUE;
 }
 
-
-
 /*************************************************
 *    Deliver waiting message down same socket    *
 *************************************************/
@@ -1534,8 +1933,8 @@ Returns:          FALSE if fork fails; TRUE otherwise
 */
 
 BOOL
-transport_pass_socket(uschar *transport_name, uschar *hostname,
-  uschar *hostaddress, uschar *id, int socket_fd)
+transport_pass_socket(const uschar *transport_name, const uschar *hostname,
+  const uschar *hostaddress, uschar *id, int socket_fd)
 {
 pid_t pid;
 int status;
@@ -1544,8 +1943,8 @@ DEBUG(D_transport) debug_printf("transport_pass_socket entered\n");
 
 if ((pid = fork()) == 0)
   {
-  int i = 16;
-  uschar **argv;
+  int i = 17;
+  const uschar **argv;
 
   /* Disconnect entirely from the parent process. If we are running in the
   test harness, wait for a bit to allow the previous process time to finish,
@@ -1553,21 +1952,22 @@ if ((pid = fork()) == 0)
   automatic comparison. */
 
   if ((pid = fork()) != 0) _exit(EXIT_SUCCESS);
-  if (running_in_test_harness) millisleep(500);
+  if (running_in_test_harness) sleep(1);
 
   /* Set up the calling arguments; use the standard function for the basics,
   but we have a number of extras that may be added. */
 
-  argv = child_exec_exim(CEE_RETURN_ARGV, TRUE, &i, FALSE, 0);
+  argv = CUSS child_exec_exim(CEE_RETURN_ARGV, TRUE, &i, FALSE, 0);
 
   if (smtp_authenticated) argv[i++] = US"-MCA";
 
-  #ifdef SUPPORT_TLS
-  if (tls_offered) argv[i++] = US"-MCT";
-  #endif
-
-  if (smtp_use_size) argv[i++] = US"-MCS";
-  if (smtp_use_pipelining) argv[i++] = US"-MCP";
+  if (smtp_peer_options & PEER_OFFERED_CHUNKING) argv[i++] = US"-MCK";
+  if (smtp_peer_options & PEER_OFFERED_DSN) argv[i++] = US"-MCD";
+  if (smtp_peer_options & PEER_OFFERED_PIPE) argv[i++] = US"-MCP";
+  if (smtp_peer_options & PEER_OFFERED_SIZE) argv[i++] = US"-MCS";
+#ifdef SUPPORT_TLS
+  if (smtp_peer_options & PEER_OFFERED_TLS) argv[i++] = US"-MCT";
+#endif
 
   if (queue_run_pid != (pid_t)0)
     {
@@ -1577,9 +1977,9 @@ if ((pid = fork()) == 0)
     }
 
   argv[i++] = US"-MC";
-  argv[i++] = transport_name;
-  argv[i++] = hostname;
-  argv[i++] = hostaddress;
+  argv[i++] = US transport_name;
+  argv[i++] = US hostname;
+  argv[i++] = US hostaddress;
   argv[i++] = string_sprintf("%d", continue_sequence + 1);
   argv[i++] = id;
   argv[i++] = NULL;
@@ -1588,8 +1988,8 @@ if ((pid = fork()) == 0)
 
   if (socket_fd != 0)
     {
-    dup2(socket_fd, 0);
-    close(socket_fd);
+    (void)dup2(socket_fd, 0);
+    (void)close(socket_fd);
     }
 
   DEBUG(D_exec) debug_print_argv(argv);
@@ -1633,7 +2033,7 @@ case, no addresses are passed.
 
 Arguments:
   argvptr            pointer to anchor for argv vector
-  cmd                points to the command string
+  cmd                points to the command string (modified IN PLACE)
   expand_arguments   true if expansion is to occur
   expand_failed      error value to set if expansion fails; not relevant if
                      addr == NULL
@@ -1647,11 +2047,12 @@ Returns:             TRUE if all went well; otherwise an error will be
 */
 
 BOOL
-transport_set_up_command(uschar ***argvptr, uschar *cmd, BOOL expand_arguments,
-  int expand_failed, address_item *addr, uschar *etext, uschar **errptr)
+transport_set_up_command(const uschar ***argvptr, uschar *cmd,
+  BOOL expand_arguments, int expand_failed, address_item *addr,
+  uschar *etext, uschar **errptr)
 {
 address_item *ad;
-uschar **argv;
+const uschar **argv;
 uschar *s, *ss;
 int address_count = 0;
 int argcount = 0;
@@ -1685,7 +2086,7 @@ while (*s != 0 && argcount < max_args)
     if (*s != 0) s++;
     *ss++ = 0;
     }
-  else argv[argcount++] = string_dequote(&s);
+  else argv[argcount++] = string_copy(string_dequote(CUSS &s));
   while (isspace(*s)) s++;
   }
 
@@ -1759,7 +2160,123 @@ if (expand_arguments)
         memmove(argv + i + 1 + additional, argv + i + 1,
           (argcount - i)*sizeof(uschar *));
 
-      for (ad = addr; ad != NULL; ad = ad->next) argv[i++] = ad->address;
+      for (ad = addr; ad != NULL; ad = ad->next) {
+          argv[i++] = ad->address;
+          argcount++;
+      }
+
+      /* Subtract one since we replace $pipe_addresses */
+      argcount--;
+      i--;
+      }
+
+      /* Handle special case of $address_pipe when af_force_command is set */
+
+    else if (addr != NULL && testflag(addr,af_force_command) &&
+        (Ustrcmp(argv[i], "$address_pipe") == 0 ||
+         Ustrcmp(argv[i], "${address_pipe}") == 0))
+      {
+      int address_pipe_i;
+      int address_pipe_argcount = 0;
+      int address_pipe_max_args;
+      uschar **address_pipe_argv;
+
+      /* We can never have more then the argv we will be loading into */
+      address_pipe_max_args = max_args - argcount + 1;
+
+      DEBUG(D_transport)
+        debug_printf("address_pipe_max_args=%d\n", address_pipe_max_args);
+
+      /* We allocate an additional for (uschar *)0 */
+      address_pipe_argv = store_get((address_pipe_max_args+1)*sizeof(uschar *));
+
+      /* +1 because addr->local_part[0] == '|' since af_force_command is set */
+      s = expand_string(addr->local_part + 1);
+
+      if (s == NULL || *s == '\0')
+        {
+        addr->transport_return = FAIL;
+        addr->message = string_sprintf("Expansion of \"%s\" "
+           "from command \"%s\" in %s failed: %s",
+           (addr->local_part + 1), cmd, etext, expand_string_message);
+        return FALSE;
+        }
+
+      while (isspace(*s)) s++; /* strip leading space */
+
+      while (*s != 0 && address_pipe_argcount < address_pipe_max_args)
+        {
+        if (*s == '\'')
+          {
+          ss = s + 1;
+          while (*ss != 0 && *ss != '\'') ss++;
+          address_pipe_argv[address_pipe_argcount++] = ss = store_get(ss - s++);
+          while (*s != 0 && *s != '\'') *ss++ = *s++;
+          if (*s != 0) s++;
+          *ss++ = 0;
+          }
+        else address_pipe_argv[address_pipe_argcount++] =
+             string_copy(string_dequote(CUSS &s));
+        while (isspace(*s)) s++; /* strip space after arg */
+        }
+
+      address_pipe_argv[address_pipe_argcount] = (uschar *)0;
+
+      /* If *s != 0 we have run out of argument slots. */
+      if (*s != 0)
+        {
+        uschar *msg = string_sprintf("Too many arguments in $address_pipe "
+          "\"%s\" in %s", addr->local_part + 1, etext);
+        if (addr != NULL)
+          {
+          addr->transport_return = FAIL;
+          addr->message = msg;
+          }
+        else *errptr = msg;
+        return FALSE;
+        }
+
+      /* address_pipe_argcount - 1
+       * because we are replacing $address_pipe in the argument list
+       * with the first thing it expands to */
+      if (argcount + address_pipe_argcount - 1 > max_args)
+        {
+        addr->transport_return = FAIL;
+        addr->message = string_sprintf("Too many arguments to command "
+          "\"%s\" after expanding $address_pipe in %s", cmd, etext);
+        return FALSE;
+        }
+
+      /* If we are not just able to replace the slot that contained
+       * $address_pipe (address_pipe_argcount == 1)
+       * We have to move the existing argv by address_pipe_argcount - 1
+       * Visually if address_pipe_argcount == 2:
+       * [argv 0][argv 1][argv 2($address_pipe)][argv 3][0]
+       * [argv 0][argv 1][ap_arg0][ap_arg1][old argv 3][0]
+       */
+      if (address_pipe_argcount > 1)
+        memmove(
+          /* current position + additonal args */
+          argv + i + address_pipe_argcount,
+          /* current position + 1 (for the (uschar *)0 at the end) */
+          argv + i + 1,
+          /* -1 for the (uschar *)0 at the end)*/
+          (argcount - i)*sizeof(uschar *)
+        );
+
+      /* Now we fill in the slots we just moved argv out of
+       * [argv 0][argv 1][argv 2=pipeargv[0]][argv 3=pipeargv[1]][old argv 3][0]
+       */
+      for (address_pipe_i = 0;
+           address_pipe_argv[address_pipe_i] != (uschar *)0;
+           address_pipe_i++)
+        {
+        argv[i++] = address_pipe_argv[address_pipe_i];
+        argcount++;
+        }
+
+      /* Subtract one since we replace $address_pipe */
+      argcount--;
       i--;
       }
 
@@ -1767,9 +2284,9 @@ if (expand_arguments)
 
     else
       {
-      uschar *expanded_arg;
+      const uschar *expanded_arg;
       enable_dollar_recipients = allow_dollar_recipients;
-      expanded_arg = expand_string(argv[i]);
+      expanded_arg = expand_cstring(argv[i]);
       enable_dollar_recipients = FALSE;
 
       if (expanded_arg == NULL)
@@ -1800,4 +2317,6 @@ if (expand_arguments)
 return TRUE;
 }
 
+/* vi: aw ai sw=2
+*/
 /* End of transport.c */