WIP: Acknowledge received commandsHEADmaster

Event the smallest protocol change has great consequences.

Relay events have been reordered to improve forward compatibility.

WIP:
 - xM handling (send)
 - xA handling (relaySend)
 - xC: fix connection killing

1 files changed, 32 insertions, 10 deletions

diff --git a/xC.c b/xC.c
index 73ddd12..1ce3d8d 100644
--- a/xC.c
+++ b/xC.c
@@ -1818,6 +1818,7 @@ struct client
 
 	uint32_t event_seq;                 ///< Outgoing message counter
 	bool initialized;                   ///< Initial sync took place
+	bool shutdown;                      ///< Shutting down
 
 	struct poller_fd socket_event;      ///< The socket can be read/written to
 };
@@ -4168,9 +4169,7 @@ relay_send (struct client *c)
 {
 	struct relay_event_message *m = &c->ctx->relay_message;
 	m->event_seq = c->event_seq++;
-
-	// TODO: Also don't try sending anything if half-closed.
-	if (!c->initialized || c->socket_fd == -1)
+	if (!c->initialized || c->shutdown || c->socket_fd == -1)
 		return;
 
 	// liberty has msg_{reader,writer} already, but they use 8-byte lengths.
@@ -4180,7 +4179,10 @@ relay_send (struct client *c)
 	 || (frame_len = c->write_buffer.len - frame_len_pos - 4) > UINT32_MAX)
 	{
 		print_error ("serialization failed, killing client");
-		client_kill (c);
+		// FIXME: This must not be done immediately!
+		//client_kill (c);
+		// TODO: Perhaps set an idle task that collects shutdown clients.
+		c->shutdown = true;
 		return;
 	}
 
@@ -15716,26 +15718,31 @@ client_process_message (struct client *c,
 		return true;
 	}
 
+	bool acknowledge = false;
 	switch (m->data.command)
 	{
 	case RELAY_COMMAND_HELLO:
+		c->initialized = true;
 		if (m->data.hello.version != RELAY_VERSION)
 		{
-			// TODO: This should send back an error message and shut down.
 			log_global_error (c->ctx,
 				"Protocol version mismatch, killing client");
-			return false;
+			relay_prepare_error (c->ctx,
+				m->command_seq, "Protocol version mismatch");
+			relay_send (c);
+
+			c->shutdown = true;
+			return true;
 		}
-		c->initialized = true;
 		client_resync (c);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_PING:
-		relay_prepare_response (c->ctx, m->command_seq)
-			->data.command = RELAY_COMMAND_PING;
-		relay_send (c);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_ACTIVE:
 		reset_autoaway (c->ctx);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_BUFFER_COMPLETE:
 		client_process_buffer_complete (c, m->command_seq, buffer,
@@ -15743,12 +15750,15 @@ client_process_message (struct client *c,
 		break;
 	case RELAY_COMMAND_BUFFER_ACTIVATE:
 		buffer_activate (c->ctx, buffer);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_BUFFER_INPUT:
 		client_process_buffer_input (c, buffer, m->data.buffer_input.text.str);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_BUFFER_TOGGLE_UNIMPORTANT:
 		buffer_toggle_unimportant (c->ctx, buffer);
+		acknowledge = true;
 		break;
 	case RELAY_COMMAND_BUFFER_LOG:
 		client_process_buffer_log (c, m->command_seq, buffer);
@@ -15758,6 +15768,12 @@ client_process_message (struct client *c,
 		relay_prepare_error (c->ctx, m->command_seq, "Unknown command");
 		relay_send (c);
 	}
+	if (acknowledge)
+	{
+		relay_prepare_response (c->ctx, m->command_seq)
+			->data.command = m->data.command;
+		relay_send (c);
+	}
 	return true;
 }
 
@@ -15851,7 +15867,13 @@ on_client_ready (const struct pollfd *pfd, void *user_data)
 {
 	struct client *c = user_data;
 	if (client_try_read (c) && client_try_write (c))
+	{
 		client_update_poller (c, pfd);
+
+		// There must be something in the write buffer if you request shutdown.
+		if (c->shutdown && !c->write_buffer.len)
+			client_kill (c);
+	}
 }
 
 // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -