abstract: existing connection compare

7 files changed, 830 insertions, 278 deletions

diff --git a/lib/abstract/abstract.c b/lib/abstract/abstract.c
index 9f40b825..cbf3db2f 100644
--- a/lib/abstract/abstract.c
+++ b/lib/abstract/abstract.c
@@ -30,18 +30,23 @@ extern const lws_abs_transport_t lws_abs_transport_cli_raw_skt,
 #if defined(LWS_WITH_SMTP)
 extern const lws_abs_protocol_t lws_abs_protocol_smtp;
 #endif
+#if defined(LWS_WITH_MQTT)
+extern const lws_abs_protocol_t lws_abs_protocol_mqttc;
+#endif
 
 static const lws_abs_transport_t * const available_abs_transports[] = {
 	&lws_abs_transport_cli_raw_skt,
 	&lws_abs_transport_cli_unit_test,
 };
 
-/* HACK: microsoft compiler can't handle zero length array definition */
-#if defined(LWS_WITH_SMTP)
+#if defined(LWS_WITH_ABSTRACT)
 static const lws_abs_protocol_t * const available_abs_protocols[] = {
 #if defined(LWS_WITH_SMTP)
 	&lws_abs_protocol_smtp,
 #endif
+#if defined(LWS_WITH_MQTT)
+	&lws_abs_protocol_mqttc,
+#endif
 };
 #endif
 
@@ -62,7 +67,7 @@ lws_abs_transport_get_by_name(const char *name)
 const lws_abs_protocol_t *
 lws_abs_protocol_get_by_name(const char *name)
 {
-#if defined(LWS_WITH_SMTP)
+#if defined(LWS_WITH_ABSTRACT)
 	int n;
 
 	for (n = 0; n < (int)LWS_ARRAY_SIZE(available_abs_protocols); n++)
@@ -89,20 +94,59 @@ lws_abs_get_token(const lws_token_map_t *token_map, short name_index)
 	return NULL;
 }
 
-void
-lws_abs_destroy_instance(lws_abs_t **ai)
+static int
+lws_abstract_compare_connection(lws_abs_t *abs1, lws_abs_t *abs2)
 {
-	lws_abs_t *a = *ai;
+	/* it has to be using the same protocol */
+	if (abs1->ap != abs2->ap)
+		return 1;
 
-	if (a->api)
-		a->ap->destroy(&a->api);
-	if (a->ati)
-		a->at->destroy(&a->ati);
+	/* protocol has to allow some kind of binding */
+	if (!abs1->ap->flags)
+		return 1;
 
-	lws_dll2_remove(&a->abstract_instances);
+	/* it has to be using the same transport */
+	if (abs1->at != abs2->at)
+		return 1;
 
-	*ai = NULL;
-	free(a);
+	/*
+	 * The transport must feel the endpoint and conditions in use match the
+	 * requested endpoint and conditions... and the transport type must be
+	 * willing to allow it
+	 */
+	if (abs1->at->compare(abs1, abs2))
+		return 1;
+
+	/*
+	 * The protocol must feel they are in compatible modes if any
+	 * (and the protocol type must be willing to allow it)
+	 */
+	if (abs1->ap->compare(abs1, abs2))
+		return 1;
+
+	/*
+	 * If no objection by now, we can say there's already a comparable
+	 * connection and both the protocol and transport feel we can make
+	 * use of it.
+	 */
+
+	return 0;
+}
+
+static int
+find_compatible(struct lws_dll2 *d, void *user)
+{
+	lws_abs_t *ai1 = (lws_abs_t *)user,
+		  *ai2 = lws_container_of(d, lws_abs_t, abstract_instances);
+
+	if (!lws_abstract_compare_connection(ai1, ai2)) {
+		/* we can bind to it */
+		lws_dll2_add_tail(&ai1->bound, &ai2->children_owner);
+
+		return 1;
+	}
+
+	return 0;
 }
 
 lws_abs_t *
@@ -110,6 +154,7 @@ lws_abs_bind_and_create_instance(const lws_abs_t *abs)
 {
 	size_t size = sizeof(lws_abs_t) + abs->ap->alloc + abs->at->alloc;
 	lws_abs_t *ai;
+	int n;
 
 	/*
 	 * since we know we will allocate the lws_abs_t, the protocol's
@@ -124,10 +169,27 @@ lws_abs_bind_and_create_instance(const lws_abs_t *abs)
 	ai->ati = NULL;
 
 	ai->api = (char *)ai + sizeof(lws_abs_t);
-	if (ai->ap->create(ai)) {
-		ai->api = NULL;
-		goto bail;
-	}
+
+	if (!ai->ap->flags) /* protocol only understands single connections */
+		goto fresh;
+
+	lws_vhost_lock(ai->vh); /* ----------------------------------- vh { */
+
+	/*
+	 * Let's have a look for any already-connected transport we can use
+	 */
+
+	n = lws_dll2_foreach_safe(&ai->vh->abstract_instances_owner, ai,
+				  find_compatible);
+
+	lws_vhost_unlock(ai->vh); /* } vh --------------------------------- */
+
+	if (n)
+		goto vh_list_add;
+
+	/* there's no existing connection doing what we want */
+
+fresh:
 
 	ai->ati = (char *)ai->api + abs->ap->alloc;
 	if (ai->at->create(ai)) {
@@ -135,12 +197,36 @@ lws_abs_bind_and_create_instance(const lws_abs_t *abs)
 		goto bail;
 	}
 
+vh_list_add:
 	/* add us to the vhost's dll2 of instances */
 
 	lws_dll2_clear(&ai->abstract_instances);
 	lws_dll2_add_head(&ai->abstract_instances,
 			  &ai->vh->abstract_instances_owner);
 
+	if (ai->ap->create(ai)) {
+		ai->api = NULL;
+		goto bail;
+	}
+
+	if (ai->bound.owner) { /* we are a piggybacker */
+		lws_abs_t *ai2 = lws_container_of(ai->bound.owner, lws_abs_t,
+						  children_owner);
+		/*
+		 * Provide an 'event' in the parent context to start handling
+		 * the bind if it's otherwise idle.  We give the parent abs
+		 * because we don't know if we're "next" or whatever.  Just that
+		 * a child joined him and he should look into his child
+		 * situation in case he was waiting for one to appear.
+		 */
+		if (ai2->ap->child_bind(ai2)) {
+			lwsl_info("%s: anticpated child bind fail\n", __func__);
+			lws_dll2_remove(&ai->bound);
+
+			goto bail;
+		}
+	}
+
 	return ai;
 
 bail:
@@ -148,3 +234,122 @@ bail:
 
 	return NULL;
 }
+
+/*
+ * We get called to clean up each child that was still bound to a parent
+ * at the time the parent is getting destroyed.
+ */
+
+static void
+__lws_abs_destroy_instance2(lws_abs_t **ai)
+{
+	lws_abs_t *a = *ai;
+
+	if (a->api)
+		a->ap->destroy(&a->api);
+	if (a->ati)
+		a->at->destroy(&a->ati);
+
+	lws_dll2_remove(&a->abstract_instances);
+
+	*ai = NULL;
+	free(a);
+}
+
+static int
+__reap_children(struct lws_dll2 *d, void *user)
+{
+	lws_abs_t *ac = lws_container_of(d, lws_abs_t, bound);
+
+	lws_dll2_foreach_safe(&ac->children_owner, NULL, __reap_children);
+
+	/* then destroy ourselves */
+
+	__lws_abs_destroy_instance2(&ac);
+
+	return 0;
+}
+
+void
+lws_abs_destroy_instance(lws_abs_t **ai)
+{
+	lws_abs_t *a = *ai;
+
+	/* destroy child instances that are bound to us first... */
+
+	lws_vhost_lock(a->vh); /* ----------------------------------- vh { */
+
+	lws_dll2_foreach_safe(&a->children_owner, NULL, __reap_children);
+
+	/* ...then destroy ourselves */
+
+	__lws_abs_destroy_instance2(ai);
+
+	lws_vhost_unlock(a->vh); /* } vh --------------------------------- */
+}
+
+lws_abs_t *
+lws_abstract_alloc(struct lws_vhost *vhost, void *user,
+		   const char *abstract_path, const lws_token_map_t *ap_tokens,
+		   const lws_token_map_t *at_tokens, struct lws_sequencer *seq,
+		   void *opaque_user_data)
+{
+	lws_abs_t *abs = lws_zalloc(sizeof(*abs), __func__);
+	struct lws_tokenize ts;
+	lws_tokenize_elem e;
+	char tmp[30];
+
+	if (!abs)
+		return NULL;
+
+	lws_tokenize_init(&ts, abstract_path, LWS_TOKENIZE_F_MINUS_NONTERM);
+
+	e = lws_tokenize(&ts);
+	if (e != LWS_TOKZE_TOKEN)
+		goto abs_path_problem;
+
+	if (lws_tokenize_cstr(&ts, tmp, sizeof(tmp)))
+		goto abs_path_problem;
+
+	abs->ap = lws_abs_protocol_get_by_name(tmp);
+	if (!abs->ap)
+		goto abs_path_problem;
+
+	e = lws_tokenize(&ts);
+	if (e != LWS_TOKZE_DELIMITER)
+		goto abs_path_problem;
+
+	e = lws_tokenize(&ts);
+	if (e != LWS_TOKZE_TOKEN)
+		goto abs_path_problem;
+
+	if (lws_tokenize_cstr(&ts, tmp, sizeof(tmp)))
+		goto abs_path_problem;
+
+	abs->at = lws_abs_transport_get_by_name(tmp);
+	if (!abs->at)
+		goto abs_path_problem;
+
+	abs->vh = vhost;
+	abs->ap_tokens = ap_tokens;
+	abs->at_tokens = at_tokens;
+	abs->seq = seq;
+	abs->opaque_user_data = opaque_user_data;
+
+	lwsl_info("%s: allocated %s\n", __func__, abstract_path);
+
+	return abs;
+
+abs_path_problem:
+	lwsl_err("%s: bad abs path '%s'\n", __func__, abstract_path);
+	lws_free_set_NULL(abs);
+
+	return NULL;
+}
+
+void
+lws_abstract_free(lws_abs_t **pabs)
+{
+	if (*pabs)
+		lws_free_set_NULL(*pabs);
+}
diff --git a/lib/abstract/private-lib-abstract.h b/lib/abstract/private-lib-abstract.h
index 90afa298..0204b503 100644
--- a/lib/abstract/private-lib-abstract.h
+++ b/lib/abstract/private-lib-abstract.h
@@ -22,4 +22,34 @@
  * IN THE SOFTWARE.
  */
 
+#if !defined(__PRIVATE_LIB_ABSTRACT_H__)
+#define __PRIVATE_LIB_ABSTRACT_H__
+
+typedef struct lws_token_map lws_token_map_t;
+typedef void lws_abs_transport_inst_t;
+typedef void lws_abs_protocol_inst_t;
+
+typedef struct lws_abs {
+	void				*user;
+	struct lws_vhost		*vh;
+
+	const struct lws_abs_protocol	*ap;
+	const lws_token_map_t		*ap_tokens;
+	const struct lws_abs_transport	*at;
+	const lws_token_map_t		*at_tokens;
+
+	struct lws_sequencer		*seq;
+	void				*opaque_user_data;
+
+	/* vh lock */
+	struct lws_dll2_owner		children_owner; /* our children / queue */
+	/* vh lock */
+	struct lws_dll2			bound; /* parent or encapsulator */
+	/* vh lock */
+	struct lws_dll2			abstract_instances;
+	lws_abs_transport_inst_t	*ati;
+	lws_abs_protocol_inst_t		*api;
+} lws_abs_t;
+
+#endif
 
diff --git a/lib/abstract/protocols/smtp/private-lib-abstract-protocols-smtp.h b/lib/abstract/protocols/smtp/private-lib-abstract-protocols-smtp.h
new file mode 100644
index 00000000..0498e08f
--- /dev/null
+++ b/lib/abstract/protocols/smtp/private-lib-abstract-protocols-smtp.h
@@ -0,0 +1,24 @@
+#define LWS_SMTP_MAX_EMAIL_LEN 32
+
+
+/*
+ * These are allocated on to the heap with an over-allocation to hold the
+ * payload at the end
+ */
+
+typedef struct lws_smtp_email {
+	struct lws_dll2	list;
+	void		*data;
+
+	char		from[LWS_SMTP_MAX_EMAIL_LEN];
+	char		to[LWS_SMTP_MAX_EMAIL_LEN];
+
+	time_t		added;
+	time_t		last_try;
+
+	lws_smtp_cb_t	done;
+
+	int		tries;
+
+	/* email payload follows */
+} lws_smtp_email_t;
diff --git a/lib/abstract/protocols/smtp/smtp-sequencer.c b/lib/abstract/protocols/smtp/smtp-sequencer.c
new file mode 100644
index 00000000..26751c5d
--- /dev/null
+++ b/lib/abstract/protocols/smtp/smtp-sequencer.c
@@ -0,0 +1,320 @@
+/*
+ * Abstract SMTP support for libwebsockets - SMTP sequencer
+ *
+ * Copyright (C) 2016-2019 Andy Green <andy@warmcat.com>
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation:
+ * version 2.1 of the License.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
+ * MA  02110-1301  USA
+ *
+ * This sequencer sits above the abstract protocol, and manages queueing,
+ * retrying mail transmission, and retry limits.
+ *
+ * Having the sequencer means that, eg, we can manage retries after complete
+ * connection failure.
+ *
+ * Connections to the smtp server are serialized
+ */
+
+#include "private-lib-core.h"
+#include "private-lib-abstract-protocols-smtp.h"
+
+typedef enum {
+	LSMTPSS_DISCONNECTED,
+	LSMTPSS_CONNECTING,
+	LSMTPSS_CONNECTED,
+	LSMTPSS_BUSY,
+} smtpss_connstate_t;
+
+typedef struct lws_smtp_sequencer {
+	struct lws_dll2_owner		emails_owner; /* email queue */
+
+	lws_abs_t			*abs, *instance;
+	lws_smtp_sequencer_args_t	args;
+	struct lws_sequencer		*seq;
+
+	smtpss_connstate_t		connstate;
+
+	time_t				email_connect_started;
+
+	/* holds the HELO for the smtp protocol to consume */
+	lws_token_map_t			apt[3];
+} lws_smtp_sequencer_t;
+
+/* sequencer messages specific to this sequencer */
+
+enum {
+	SEQ_MSG_CLIENT_FAILED = LWSSEQ_USER_BASE,
+	SEQ_MSG_CLIENT_DONE,
+};
+
+/*
+ * We're going to bind to the raw-skt transport, so tell that what we want it
+ * to connect to
+ */
+
+static const lws_token_map_t smtp_rs_transport_tokens[] = {
+ {
+	.u = { .value = "127.0.0.1" },
+	.name_index = LTMI_PEER_V_DNS_ADDRESS,
+ }, {
+	.u = { .lvalue = 25 },
+	.name_index = LTMI_PEER_LV_PORT,
+ }, {
+ }
+};
+
+static void
+lws_smtpc_kick_internal(lws_smtp_sequencer_t *s)
+{
+	lws_smtp_email_t *e;
+	lws_dll2_t *d;
+	char buf[64];
+	int n;
+	lws_dll2_t *pd2;
+
+	pd2 = lws_dll2_get_head(&s->emails_owner);
+	if (!pd2)
+		return;
+
+	e = lws_container_of(pd2, lws_smtp_email_t, list);
+	if (!e)
+		return;
+
+	/* Is there something to do?  If so, we need a connection... */
+
+	if (s->connstate == LSMTPSS_DISCONNECTED) {
+
+		s->apt[0].u.value = s->args.helo;
+		s->apt[0].name_index = LTMI_PSMTP_V_HELO;
+		s->apt[1].u.value = (void *)e;
+		s->apt[1].name_index = LTMI_PSMTP_V_LWS_SMTP_EMAIL_T;
+
+		/*
+		 * create and connect the smtp protocol + transport
+		 */
+
+		s->abs = lws_abstract_alloc(s->args.vhost, NULL, "smtp.raw_skt",
+					    s->apt, smtp_rs_transport_tokens,
+					    s->seq, NULL);
+		if (!s->abs)
+			return;
+
+		s->instance = lws_abs_bind_and_create_instance(s->abs);
+		if (!s->instance) {
+			lws_abstract_free(&s->abs);
+			lwsl_err("%s: failed to create SMTP client\n", __func__);
+
+			goto bail1;
+		}
+
+		s->connstate = LSMTPSS_CONNECTING;
+		lws_sequencer_timeout(s->seq, 10);
+		return;
+	}
+
+	/* ask the transport if we have a connection to the server ongoing */
+
+	if (s->abs->at->state(s->abs->ati)) {
+		/*
+		 * there's a connection, it could be still trying to connect
+		 * or established
+		 */
+		s->abs->at->ask_for_writeable(s->abs->ati);
+
+		return;
+	}
+
+	/* there's no existing connection */
+
+	lws_smtpc_state_transition(c, LGSSMTP_CONNECTING);
+
+	if (s->abs->at->client_conn(s->abs)) {
+		lwsl_err("%s: failed to connect\n", __func__);
+
+		return;
+	}
+
+	e->tries++;
+	e->last_try = lws_now_secs();
+}
+
+
+/*
+ * The callback we get from the smtp protocol... we use this to drive
+ * decisions about destroy email, retry and fail.
+ *
+ * Sequencer will handle it via the event loop.
+ */
+
+static int
+email_result(void *e, void *d, int disp, void *b, size_t l)
+{
+	lws_smtp_sequencer_t *s = (lws_smtp_sequencer_t *)d;
+
+	lws_sequencer_event(s->seq, LWSSEQ_USER_BASE + disp, e);
+
+	return 0;
+}
+
+static int
+cleanup(struct lws_dll2 *d, void *user)
+{
+	lws_smtp_email_t *e;
+
+	e = lws_container_of(d, lws_smtp_email_t, list);
+	if (e->done)
+		e->done(e, "destroying", 10);
+
+	lws_dll2_remove(d);
+	lws_free(e);
+
+	return 0;
+}
+
+static lws_seq_cb_return_t
+smtp_sequencer_cb(struct lws_sequencer *seq, void *user, int event, void *data)
+{
+	struct lws_smtp_sequencer_t *s = (struct lws_smtp_sequencer_t *)user;
+
+	switch ((int)event) {
+	case LWSSEQ_CREATED: /* our sequencer just got started */
+		lwsl_notice("%s: %s: created\n", __func__,
+			    lws_sequencer_name(seq));
+		s->connstate = LSMTPSS_DISCONNECTED;
+		s->state = 0;  /* first thing we'll do is the first url */
+		goto step;
+
+	case LWSSEQ_DESTROYED:
+		lws_dll2_foreach_safe(&s->pending_owner, NULL, cleanup);
+		break;
+
+	case LWSSEQ_TIMED_OUT:
+		lwsl_notice("%s: LWSSEQ_TIMED_OUT\n", __func__);
+		break;
+
+	case LWSSEQ_USER_BASE + LWS_SMTP_DISPOSITION_SENT:
+		lws_smtpc_free_email(data);
+		break;
+
+	case LWSSEQ_WSI_CONNECTED:
+		s->connstate = LSMTPSS_CONNECTED;
+		lws_smtpc_kick_internal(s);
+		break;
+
+	case LWSSEQ_WSI_CONN_FAIL:
+	case LWSSEQ_WSI_CONN_CLOSE:
+		s->connstate = LSMTPSS_DISCONNECTED;
+		lws_smtpc_kick_internal(s);
+		break;
+
+	case SEQ_MSG_SENT:
+		break;
+
+	default:
+		break;
+	}
+
+	return LWSSEQ_RET_CONTINUE;
+}
+
+/*
+ * Creates an lws_sequencer to manage the test sequence
+ */
+
+lws_smtp_sequencer_t *
+lws_smtp_sequencer_create(const lws_smtp_sequencer_args_t *args)
+{
+	lws_smtp_sequencer_t *s;
+	struct lws_sequencer *seq;
+
+	/*
+	 * Create a sequencer in the event loop to manage the SMTP queue
+	 */
+
+	seq = lws_sequencer_create(args->vhost->context, 0,
+				   sizeof(lws_smtp_sequencer_t), (void **)&s,
+				   smtp_sequencer_cb, "smtp-seq");
+	if (!seq) {
+		lwsl_err("%s: unable to create sequencer\n", __func__);
+		return NULL;
+	}
+
+	s->abs = *args->abs;
+	s->args = *args;
+	s->seq = seq;
+
+	/* set defaults in our copy of the args */
+
+	if (!s->args.helo[0])
+		strcpy(s->args.helo, "default-helo");
+	if (!s->args.email_queue_max)
+		s->args.email_queue_max = 8;
+	if (!s->args.retry_interval)
+		s->args.retry_interval = 15 * 60;
+	if (!s->args.delivery_timeout)
+		s->args.delivery_timeout = 12 * 60 * 60;
+
+	return s;
+}
+
+void
+lws_smtp_sequencer_destroy(lws_smtp_sequencer_t *s)
+{
+	/* sequencer destruction destroys all assets */
+	lws_sequencer_destroy(&s->seq);
+}
+
+int
+lws_smtpc_add_email(lws_smtp_sequencer_t *s, const char *payload,
+		    size_t payload_len, const char *sender,
+		    const char *recipient, void *data, lws_smtp_cb_t done)
+{
+	lws_smtp_email_t *e;
+
+	if (s->emails_owner.count > s->args.email_queue_max) {
+		lwsl_err("%s: email queue at limit of %d\n", __func__,
+			 (int)s->args.email_queue_max);
+
+		return 1;
+	}
+
+	if (!done)
+		return 1;
+
+	e = malloc(sizeof(*e) + payload_len + 1);
+	if (!e)
+		return 1;
+
+	memset(e, 0, sizeof(*e));
+
+	e->data = data;
+	e->done = done;
+
+	lws_strncpy(e->from, sender, sizeof(e->from));
+	lws_strncpy(e->to, recipient, sizeof(e->to));
+
+	memcpy((char *)&e[1], payload, payload_len + 1);
+
+	e->added = lws_now_secs();
+	e->last_try = 0;
+	e->tries = 0;
+
+	lws_dll2_clear(&e->list);
+	lws_dll2_add_tail(&e->list, &s->emails_owner);
+
+	lws_smtpc_kick_internal(s);
+
+	return 0;
+}
diff --git a/lib/abstract/protocols/smtp/smtp.c b/lib/abstract/protocols/smtp/smtp.c
index 1d7d1542..78adf1a6 100644
--- a/lib/abstract/protocols/smtp/smtp.c
+++ b/lib/abstract/protocols/smtp/smtp.c
@@ -30,33 +30,36 @@ typedef enum lwsgs_smtp_states {
 	LGSSMTP_IDLE,		/**< awaiting new email */
 	LGSSMTP_CONNECTING,	/**< opening tcp connection to MTA */
 	LGSSMTP_CONNECTED,	/**< tcp connection to MTA is connected */
+		/* (server sends greeting) */
 	LGSSMTP_SENT_HELO,	/**< sent the HELO */
+
 	LGSSMTP_SENT_FROM,	/**< sent FROM */
 	LGSSMTP_SENT_TO,	/**< sent TO */
 	LGSSMTP_SENT_DATA,	/**< sent DATA request */
 	LGSSMTP_SENT_BODY,	/**< sent the email body */
-	LGSSMTP_SENT_QUIT,	/**< sent the session quit */
-} lwsgs_smtp_states_t;
 
-/** struct lws_email - abstract context for performing SMTP operations */
-typedef struct lws_smtp_client {
-	struct lws_dll2_owner pending_owner;
+		/*
+		 * (server sends, eg, "250 Ok: queued as 12345")
+		 * at this point we can return to LGSSMTP_SENT_HELO and send a
+		 * new email, or continue below to QUIT, or just wait
+		 */
 
-	const struct lws_abs *abs;
+	LGSSMTP_SENT_QUIT,	/**< sent the session quit */
 
-	const char *helo;
+	/* (server sends, eg, "221 Bye" and closes the connection) */
+} lwsgs_smtp_states_t;
 
-	lwsgs_smtp_states_t estate;
-	time_t email_connect_started;
+/** abstract protocol instance data */
 
-	time_t retry_interval;
-	time_t delivery_timeout;
+typedef struct lws_smtp_client_protocol {
+	const struct lws_abs	*abs;
+	lwsgs_smtp_states_t	estate;
 
-	size_t email_queue_max;
-	size_t max_content_size;
+	lws_smtp_email_t	*e;	/* the email we are trying to send */
+	const char		*helo;
 
-	unsigned char send_pending:1;
-} lws_smtp_client_t;
+	unsigned char		send_pending:1;
+} lws_smtpcp_t;
 
 static const short retcodes[] = {
 	0,	/* idle */
@@ -71,80 +74,71 @@ static const short retcodes[] = {
 };
 
 static void
-lws_smtp_client_state_transition(lws_smtp_client_t *c, lwsgs_smtp_states_t s)
+lws_smtpc_state_transition(lws_smtpcp_t *c, lwsgs_smtp_states_t s)
 {
 	lwsl_debug("%s: cli %p: state %d -> %d\n", __func__, c, c->estate, s);
 	c->estate = s;
 }
 
-static void
-lws_smtp_client_kick_internal(lws_smtp_client_t *c)
+static lws_smtp_email_t *
+lws_smtpc_get_email(lws_smtpcp_t *c)
 {
-	lws_smtp_email_t *e;
-	lws_dll2_t *d;
-	char buf[64];
-	int n;
-
-	if (c->estate != LGSSMTP_IDLE)
-		return;
-
-	/* is there something to do? */
-
-again:
-	d = lws_dll2_get_head(&c->pending_owner);
-	if (!d)
-		return;
+	const lws_token_map_t *tm;
 
-	e = lws_container_of(d, lws_smtp_email_t, list);
+	/* ... the email we want to send */
+	tm = lws_abs_get_token(c->abs->ap_tokens, LTMI_PSMTP_V_LWS_SMTP_EMAIL_T);
+	if (!tm) {
+		assert(0);
 
-	/* do we need to time out this guy? */
+		return NULL;
+	}
 
-	if ((time_t)lws_now_secs() - e->added > (time_t)c->delivery_timeout) {
-		lwsl_err("%s: timing out email\n", __func__);
-		lws_dll2_remove(&e->list);
-		n = lws_snprintf(buf, sizeof(buf), "0 Timed out retrying send");
-		e->done(e, buf, n);
+	return (lws_smtp_email_t *)tm->u.value;
+}
 
-		if (lws_dll2_get_head(&c->pending_owner))
-			goto again;
+/*
+ * Called when something happened so that we know now the final disposition of
+ * the email send attempt, for good or ill.
+ *
+ * Inform the owner via the done callback and set up the next queued one if any.
+ *
+ * Returns nonzero if we queued a new one
+ */
 
-		return;
-	}
+static int
+lws_smtpc_email_disposition(lws_smtpcp_t *c, int disp, const void *buf,
+			    size_t len)
+{
+	lws_smtpcp_t *ch;
+	lws_abs_t *ach;
+	lws_dll2_t *d;
 
-	/* is it time for his retry yet? */
+	lws_smtpc_state_transition(c, LGSSMTP_SENT_HELO);
 
-	if (e->last_try &&
-	    (time_t)lws_now_secs() - e->last_try < (time_t)c->retry_interval) {
-		/* no... send him to the tail */
-		lws_dll2_remove(&e->list);
-		lws_dll2_add_tail(&e->list, &c->pending_owner);
-		return;
-	}
+	/* lifetime of the email object is handled by done callback */
+	c->e->done(c->e, c->e->data, disp, buf, len);
+	c->e = NULL;
 
-	/* ask the transport if we have a connection to the server ongoing */
+	/* this may not be the time to try to send anything else... */
 
-	if (c->abs->at->state(c->abs->ati)) {
-		/*
-		 * there's a connection, it could be still trying to connect
-		 * or established
-		 */
-		c->abs->at->ask_for_writeable(c->abs->ati);
+	if (disp == LWS_SMTP_DISPOSITION_FAILED_DESTROY)
+		return 0;
 
-		return;
-	}
+	/* ... otherwise... do we have another queued? */
 
-	/* there's no existing connection */
+	d = lws_dll2_get_tail(&c->abs->children_owner);
+	if (!d)
+		return 0;
 
-	lws_smtp_client_state_transition(c, LGSSMTP_CONNECTING);
+	ach = lws_container_of(d, lws_abs_t, bound);
+	ch = (lws_smtpcp_t *)ach->api;
 
-	if (c->abs->at->client_conn(c->abs)) {
-		lwsl_err("%s: failed to connect\n", __func__);
+	c->e = lws_smtpc_get_email(ch);
 
-		return;
-	}
+	/* since we took it on, remove it from the queue */
+	lws_dll2_remove(d);
 
-	e->tries++;
-	e->last_try = lws_now_secs();
+	return 1;
 }
 
 /*
@@ -152,53 +146,82 @@ again:
  */
 
 static int
-lws_smtp_client_abs_accept(lws_abs_protocol_inst_t *api)
+lws_smtpc_abs_accept(lws_abs_protocol_inst_t *api)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)api;
+	lws_smtpcp_t *c = (lws_smtpcp_t *)api;
+
+	/* we have become connected in the tcp sense */
+
+	lws_smtpc_state_transition(c, LGSSMTP_CONNECTED);
+
+	/*
+	 * From the accept(), the next thing that should happen is the SMTP
+	 * server sends its greeting like "220 smtp2.example.com ESMTP Postfix",
+	 * we'll hear about it in the rx callback, or time out
+	 */
 
-	lws_smtp_client_state_transition(c, LGSSMTP_CONNECTED);
+	c->abs->at->set_timeout(c->abs->ati,
+				PENDING_TIMEOUT_AWAITING_SERVER_RESPONSE, 3);
 
 	return 0;
 }
 
 static int
-lws_smtp_client_abs_rx(lws_abs_protocol_inst_t *api, uint8_t *buf, size_t len)
+lws_smtpc_abs_rx(lws_abs_protocol_inst_t *api, const uint8_t *buf, size_t len)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)api;
-	lws_smtp_email_t *e;
-	lws_dll2_t *pd2;
+	lws_smtpcp_t *c = (lws_smtpcp_t *)api;
+	char at[5];
 	int n;
 
-	pd2 = lws_dll2_get_head(&c->pending_owner);
-	if (!pd2)
-		return 0;
+	c->abs->at->set_timeout(c->abs->ati, NO_PENDING_TIMEOUT, 0);
 
-	e = lws_container_of(pd2, lws_smtp_email_t, list);
-	if (!e)
-		return 0;
+	lws_strncpy(at, (const char *)buf, sizeof(at));
+	n = atoi(at);
 
-	n = atoi((char *)buf);
-	if (n != retcodes[c->estate]) {
-		lwsl_notice("%s: bad response from server: %d (state %d) %.*s\n",
-				__func__, n, c->estate, (int)len, buf);
+	switch (c->estate) {
+	case LGSSMTP_CONNECTED:
+		if (n != 220) {
+			/*
+			 * The server did not properly greet us... we can't
+			 * even get started, so fail the transport connection
+			 * (and anything queued on it)
+			 */
+			lwsl_err("%s: server: %.*s\n", __func__, (int)len, buf);
 
-		lws_dll2_remove(&e->list);
-		lws_dll2_add_tail(&e->list, &c->pending_owner);
-		lws_smtp_client_state_transition(c, LGSSMTP_IDLE);
-		lws_smtp_client_kick_internal(c);
+			return 1;
+		}
+		break;
 
-		return 0;
-	}
+	case LGSSMTP_SENT_BODY:
+		/*
+		 * We finished one way or another... let's prepare to send a
+		 * new one... or wait until server hangs up on us
+		 */
+		if (!lws_smtpc_email_disposition(c,
+					n == 250 ? LWS_SMTP_DISPOSITION_SENT :
+						   LWS_SMTP_DISPOSITION_FAILED,
+					"destroyed", 0))
+			return 0; /* become idle */
 
-	if (c->estate == LGSSMTP_SENT_QUIT) {
-		lwsl_debug("%s: done\n", __func__);
-		lws_smtp_client_state_transition(c, LGSSMTP_IDLE);
+		break; /* ask to send */
 
-		lws_dll2_remove(&e->list);
-		if (e->done && e->done(e, "sent OK", 7))
-			return 1;
+	case LGSSMTP_SENT_QUIT:
+		lwsl_debug("%s: done\n", __func__);
+		lws_smtpc_state_transition(c, LGSSMTP_IDLE);
 
 		return 1;
+
+	default:
+		if (n != retcodes[c->estate]) {
+			lwsl_notice("%s: bad response: %d (state %d) %.*s\n",
+				    __func__, n, c->estate, (int)len, buf);
+
+			lws_smtpc_email_disposition(c,
+					LWS_SMTP_DISPOSITION_FAILED, buf, len);
+
+			return 0;
+		}
+		break;
 	}
 
 	c->send_pending = 1;
@@ -208,24 +231,13 @@ lws_smtp_client_abs_rx(lws_abs_protocol_inst_t *api, uint8_t *buf, size_t len)
 }
 
 static int
-lws_smtp_client_abs_writeable(lws_abs_protocol_inst_t *api, size_t budget)
+lws_smtpc_abs_writeable(lws_abs_protocol_inst_t *api, size_t budget)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)api;
 	char b[256 + LWS_PRE], *p = b + LWS_PRE;
-	lws_smtp_email_t *e;
-	lws_dll2_t *pd2;
+	lws_smtpcp_t *c = (lws_smtpcp_t *)api;
 	int n;
 
-	pd2 = lws_dll2_get_head(&c->pending_owner);
-	if (!pd2)
-		return 0;
-
-	e = lws_container_of(pd2, lws_smtp_email_t, list);
-	if (!e)
-		return 0;
-
-
-	if (!c->send_pending)
+	if (!c->send_pending || !c->e)
 		return 0;
 
 	c->send_pending = 0;
@@ -235,31 +247,37 @@ lws_smtp_client_abs_writeable(lws_abs_protocol_inst_t *api, size_t budget)
 	switch (c->estate) {
 	case LGSSMTP_CONNECTED:
 		n = lws_snprintf(p, sizeof(b) - LWS_PRE, "HELO %s\n", c->helo);
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_HELO);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_HELO);
 		break;
+
 	case LGSSMTP_SENT_HELO:
 		n = lws_snprintf(p, sizeof(b) - LWS_PRE, "MAIL FROM: <%s>\n",
-				 e->email_from);
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_FROM);
+				 c->e->from);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_FROM);
 		break;
+
 	case LGSSMTP_SENT_FROM:
 		n = lws_snprintf(p, sizeof(b) - LWS_PRE,
-				 "RCPT TO: <%s>\n", e->email_to);
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_TO);
+				 "RCPT TO: <%s>\n", c->e->to);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_TO);
 		break;
+
 	case LGSSMTP_SENT_TO:
 		n = lws_snprintf(p, sizeof(b) - LWS_PRE, "DATA\n");
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_DATA);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_DATA);
 		break;
+
 	case LGSSMTP_SENT_DATA:
-		p = (char *)e->payload;
-		n = strlen(e->payload);
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_BODY);
+		p = (char *)&c->e[1];
+		n = strlen(p);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_BODY);
 		break;
+
 	case LGSSMTP_SENT_BODY:
 		n = lws_snprintf(p, sizeof(b) - LWS_PRE, "quit\n");
-		lws_smtp_client_state_transition(c, LGSSMTP_SENT_QUIT);
+		lws_smtpc_state_transition(c, LGSSMTP_SENT_QUIT);
 		break;
+
 	case LGSSMTP_SENT_QUIT:
 		return 0;
 
@@ -274,179 +292,88 @@ lws_smtp_client_abs_writeable(lws_abs_protocol_inst_t *api, size_t budget)
 }
 
 static int
-lws_smtp_client_abs_closed(lws_abs_protocol_inst_t *api)
+lws_smtpc_abs_closed(lws_abs_protocol_inst_t *api)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)api;
+	lws_smtpcp_t *c = (lws_smtpcp_t *)api;
 
 	if (c)
-		lws_smtp_client_state_transition(c, LGSSMTP_IDLE);
+		lws_smtpc_state_transition(c, LGSSMTP_IDLE);
 
 	return 0;
 }
 
-static int
-lws_smtp_client_abs_heartbeat(lws_abs_protocol_inst_t *api)
-{
-	lws_smtp_client_t *c = (lws_smtp_client_t *)api;
-
-	lws_smtp_client_kick_internal(c);
-
-	return 0;
-}
+/*
+ * Creating for initial transport and for piggybacking on another transport
+ * both get created here the same.  But piggybackers have ai->bound attached.
+ */
 
-lws_smtp_email_t *
-lws_smtp_client_alloc_email_helper(const char *payload, size_t payload_len,
-				   const char *sender, const char *recipient,
-				   const char *extra, size_t extra_len, void *data,
-				   int (*done)(struct lws_smtp_email *e,
-					       void *buf, size_t len))
+static int
+lws_smtpc_create(const lws_abs_t *ai)
 {
-	size_t ls = strlen(sender), lr = strlen(recipient);
-	lws_smtp_email_t *em;
-	char *p;
+	lws_smtpcp_t *c = (lws_smtpcp_t *)ai->api;
 
-	em = malloc(sizeof(*em) + payload_len + ls + lr + extra_len + 4);
-	if (!em) {
-		lwsl_err("OOM\n");
-		return NULL;
-	}
-
-	p = (char *)&em[1];
-
-	memset(em, 0, sizeof(*em));
+	memset(c, 0, sizeof(*c));
 
-	em->data = data;
-	em->done = done;
+	c->abs = ai;
+	c->e = lws_smtpc_get_email(c);
 
-	em->email_from = p;
-	memcpy(p, sender, ls + 1);
-	p += ls + 1;
-	em->email_to = p;
-	memcpy(p, recipient, lr + 1);
-	p += lr + 1;
-	em->payload = p;
-	memcpy(p, payload, payload_len + 1);
-	p += payload_len + 1;
+	lws_smtpc_state_transition(c, lws_dll2_is_detached(&ai->bound) ?
+					LGSSMTP_CONNECTING : LGSSMTP_IDLE);
 
-	if (extra) {
-		em->extra = p;
-		memcpy(p, extra, extra_len + 1);
-	}
+	/* If we are initiating the transport, we will get an accept() next...
+	 *
+	 * If we are piggybacking, the parent will get a .child_bind() after
+	 * this to give it a chance to act on us joining (eg, it was completely
+	 * idle and we joined).
+	 */
 
-	return em;
+	return 0;
 }
 
-int
-lws_smtp_client_add_email(lws_abs_t *instance, lws_smtp_email_t *e)
+static void
+lws_smtpc_destroy(lws_abs_protocol_inst_t **_c)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)instance->api;
-
-	if (c->pending_owner.count > c->email_queue_max) {
-		lwsl_err("%s: email queue at limit of %d\n", __func__,
-				(int)c->email_queue_max);
+	lws_smtpcp_t *c = (lws_smtpcp_t *)*_c;
 
-		return 1;
-	}
-
-	e->added = lws_now_secs();
-	e->last_try = 0;
-	e->tries = 0;
-
-	lws_dll2_clear(&e->list);
-	lws_dll2_add_tail(&e->list, &c->pending_owner);
+	if (!c)
+		return;
 
-	lws_smtp_client_kick_internal(c);
+	/* so if we are still holding on to c->e, we have failed to send it */
+	if (c->e)
+		lws_smtpc_email_disposition(c,
+			LWS_SMTP_DISPOSITION_FAILED_DESTROY, "destroyed", 0);
 
-	return 0;
+	*_c = NULL;
 }
 
-void
-lws_smtp_client_kick(lws_abs_t *instance)
-{
-	lws_smtp_client_t *c = (lws_smtp_client_t *)instance->api;
-
-	lws_smtp_client_kick_internal(c);
-}
 static int
-lws_smtp_client_create(const lws_abs_t *ai)
+lws_smtpc_compare(lws_abs_t *abs1, lws_abs_t *abs2)
 {
-	lws_smtp_client_t *c = (lws_smtp_client_t *)ai->api;
-	const lws_token_map_t *tm;
-
-	memset(c, 0, sizeof(*c));
-
-	c->abs = ai;
-
-	tm = lws_abs_get_token(ai->ap_tokens, LTMI_PSMTP_V_HELO);
-	if (!tm) {
-		lwsl_err("%s: LTMI_PSMTP_V_HELO is required\n", __func__);
-
-		return 1;
-	}
-	c->helo = tm->u.value;
-
-	c->email_queue_max	= 8;
-	c->retry_interval	= 15 * 60;
-	c->delivery_timeout	= 12 * 60 * 60;
-
-	tm = lws_abs_get_token(ai->ap_tokens, LTMI_PSMTP_LV_EMAIL_QUEUE_MAX);
-	if (tm)
-		c->email_queue_max = tm->u.lvalue;
-	tm = lws_abs_get_token(ai->ap_tokens, LTMI_PSMTP_LV_RETRY_INTERVAL);
-	if (tm)
-		c->retry_interval = tm->u.lvalue;
-	tm = lws_abs_get_token(ai->ap_tokens, LTMI_PSMTP_LV_DELIVERY_TIMEOUT);
-	if (tm)
-		c->delivery_timeout = tm->u.lvalue;
-
-	lws_smtp_client_state_transition(c, LGSSMTP_IDLE);
-
 	return 0;
 }
 
 static int
-cleanup(struct lws_dll2 *d, void *user)
+lws_smtpc_child_bind(lws_abs_t *abs)
 {
-	lws_smtp_email_t *e;
-
-	e = lws_container_of(d, lws_smtp_email_t, list);
-	if (e->done && e->done(e, "destroying", 10))
-		return 1;
-
 	return 0;
 }
 
-static void
-lws_smtp_client_destroy(lws_abs_protocol_inst_t **_c)
-{
-	lws_smtp_client_t *c = (lws_smtp_client_t *)*_c;
-
-	if (!c)
-		return;
-
-	lws_dll2_foreach_safe(&c->pending_owner, NULL, cleanup);
-
-	/*
-	 * We don't free anything because the abstract layer combined our
-	 * allocation with that of the instance, and it will free the whole
-	 * thing after this.
-	 */
-
-	*_c = NULL;
-}
-
 /* events the transport invokes (handled by abstract protocol) */
 
 const lws_abs_protocol_t lws_abs_protocol_smtp = {
 	.name		= "smtp",
-	.alloc		= sizeof(lws_smtp_client_t),
+	.alloc		= sizeof(lws_smtpcp_t),
+	.flags		= LWSABSPR_FLAG_PIPELINE,
+
+	.create		= lws_smtpc_create,
+	.destroy	= lws_smtpc_destroy,
+	.compare	= lws_smtpc_compare,
 
-	.create		= lws_smtp_client_create,
-	.destroy	= lws_smtp_client_destroy,
+	.accept		= lws_smtpc_abs_accept,
+	.rx		= lws_smtpc_abs_rx,
+	.writeable	= lws_smtpc_abs_writeable,
+	.closed		= lws_smtpc_abs_closed,
+	.heartbeat	= NULL,
 
-	.accept		= lws_smtp_client_abs_accept,
-	.rx		= lws_smtp_client_abs_rx,
-	.writeable	= lws_smtp_client_abs_writeable,
-	.closed		= lws_smtp_client_abs_closed,
-	.heartbeat	= lws_smtp_client_abs_heartbeat,
+	.child_bind	= lws_smtpc_child_bind,
 };
diff --git a/lib/abstract/transports/raw-skt.c b/lib/abstract/transports/raw-skt.c
index 384807f2..fec2db73 100644
--- a/lib/abstract/transports/raw-skt.c
+++ b/lib/abstract/transports/raw-skt.c
@@ -321,9 +321,9 @@ static void
 lws_atcrs_destroy(lws_abs_transport_inst_t **pati)
 {
 	/*
-	 * We don't free anything because the abstract layer combined our
-	 * allocation with that of the instance, and it will free the whole
-	 * thing after this.
+	 * For ourselves, we don't free anything because the abstract layer
+	 * combined our allocation with that of the abs instance, and it will
+	 * free the whole thing after this.
 	 */
 	*pati = NULL;
 }
@@ -349,12 +349,51 @@ lws_atcrs_state(lws_abs_transport_inst_t *ati)
 	return 1;
 }
 
+static int
+lws_atcrs_compare(lws_abs_t *abs1, lws_abs_t *abs2)
+{
+	const lws_token_map_t *tm1, *tm2;
+
+	tm1 = lws_abs_get_token(abs1->at_tokens, LTMI_PEER_V_DNS_ADDRESS);
+	tm2 = lws_abs_get_token(abs2->at_tokens, LTMI_PEER_V_DNS_ADDRESS);
+
+	/* Address token is mandatory and must match */
+	if (!tm1 || !tm2 || strcmp(tm1->u.value, tm2->u.value))
+		return 1;
+
+	/* Port token is mandatory and must match */
+	tm1 = lws_abs_get_token(abs1->at_tokens, LTMI_PEER_LV_PORT);
+	tm2 = lws_abs_get_token(abs2->at_tokens, LTMI_PEER_LV_PORT);
+	if (!tm1 || !tm2 || tm1->u.lvalue != tm2->u.lvalue)
+		return 1;
+
+	/* TLS is optional... */
+	tm1 = lws_abs_get_token(abs1->at_tokens, LTMI_PEER_LV_TLS_FLAGS);
+	tm2 = lws_abs_get_token(abs2->at_tokens, LTMI_PEER_LV_TLS_FLAGS);
+
+	/* ... but both must have the same situation with it given or not... */
+	if (!!tm1 != !!tm2)
+		return 1;
+
+	/* if not using TLS, then that's enough to call it */
+	if (!tm1)
+		return 0;
+
+	/* ...and if there are tls flags, both must have the same tls flags */
+	if (tm1->u.lvalue != tm2->u.lvalue)
+		return 1;
+
+	/* ... and both must use the same client tls ctx / vhost */
+	return abs1->vh != abs2->vh;
+}
+
 const lws_abs_transport_t lws_abs_transport_cli_raw_skt = {
 	.name			= "raw_skt",
 	.alloc			= sizeof(abs_raw_skt_priv_t),
 
 	.create			= lws_atcrs_create,
 	.destroy		= lws_atcrs_destroy,
+	.compare		= lws_atcrs_compare,
 
 	.tx			= lws_atcrs_tx,
 #if !defined(LWS_WITH_CLIENT)
diff --git a/lib/abstract/transports/unit-test.c b/lib/abstract/transports/unit-test.c
index c5b0bc95..25052d81 100644
--- a/lib/abstract/transports/unit-test.c
+++ b/lib/abstract/transports/unit-test.c
@@ -513,12 +513,19 @@ lws_unit_test_result_name(int in)
 	return dnames[in];
 }
 
+static int
+lws_atcut_compare(lws_abs_t *abs1, lws_abs_t *abs2)
+{
+	return 0;
+}
+
 const lws_abs_transport_t lws_abs_transport_cli_unit_test = {
 	.name			= "unit_test",
 	.alloc			= sizeof(abs_unit_test_priv_t),
 
 	.create			= lws_atcut_create,
 	.destroy		= lws_atcut_destroy,
+	.compare		= lws_atcut_compare,
 
 	.tx			= lws_atcut_tx,
 #if !defined(LWS_WITH_CLIENT)