fixup! connectd: Implement sending of start_batch

Author: ddustin

connectd/multiplex.c

@@ -417,28 +417,31 @@ static u8 *process_batch_elements(struct peer *peer, const u8 *msg TAKES)
 		struct channel_id channel_id;
 		u8 *enc_msg;
 
-		if (fromwire_u16(&cursor, &plen) != WIRE_PROTOCOL_BATCH_ELEMENT)
-			status_failed(STATUS_FAIL_PROTO_BATCH,
-				      "process_batch_elements on msg that is"
+		if (fromwire_u16(&cursor, &plen) != WIRE_PROTOCOL_BATCH_ELEMENT) {
+			status_broken("process_batch_elements on msg that is"
 				      " not WIRE_PROTOCOL_BATCH_ELEMENT. %s",
 				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+		}
 
 		fromwire_channel_id(&cursor, &plen, &channel_id);
 
 	 	element_size = fromwire_u16(&cursor, &plen);
-	 	if (!element_size)
-			status_failed(STATUS_FAIL_PROTO_BATCH,
-				      "process_batch_elements cannot have zero"
+	 	if (!element_size) {
+			status_broken("process_batch_elements cannot have zero"
 				      " length elements. %s",
 				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+	 	}
 
 		element_bytes = fromwire_tal_arrn(NULL, &cursor, &plen,
 						  element_size);
-		if (!element_bytes)
-			status_failed(STATUS_FAIL_PROTO_BATCH,
-				      "process_batch_elements fromwire_tal_arrn"
+		if (!element_bytes) {
+			status_broken("process_batch_elements fromwire_tal_arrn"
 				      " %s",
 				      tal_hexstr(tmpctx, cursor, plen));
+			return tal_free(ret);
+		}
 
 		status_debug("Processing batch extracted item %s. %s",
 			     peer_wire_name(fromwire_peektype(element_bytes)),
@@ -503,10 +506,14 @@ static struct io_plan *encrypt_and_send(struct peer *peer,
 	set_urgent_flag(peer, is_urgent(type));
 
 	/* Special message type directing us to process batch items. */
-	if (type == WIRE_PROTOCOL_BATCH_ELEMENT)
+	if (type == WIRE_PROTOCOL_BATCH_ELEMENT) {
 		peer->sent_to_peer = process_batch_elements(peer, msg);
-	else
+		if (!peer->sent_to_peer)
+			return io_close(peer->to_peer);
+	}
+	else {
 		peer->sent_to_peer = cryptomsg_encrypt_msg(peer, &peer->cs, msg);
+	}
 	/* We free this and the encrypted version in next write_to_peer */
 
 	return io_write(peer->to_peer,