protocol: add query option CONTINUE

Author: MaxKellermann

debian/changelog

@@ -1,5 +1,6 @@
 cm4all-pond (0.35) unstable; urgency=low
 
+  * protocol: add query option CONTINUE
   * client: show REMOTE_HOST in message lines
   * client: support ISO8601 time stamps in message lines
   * client: show Conten-Type in JSONL response

src/Connection.cxx

@@ -24,6 +24,7 @@ Connection::Request::Clear() noexcept
 	group_site.max_sites = 0;
 	window.max = 0;
 	follow = false;
+	continue_ = false;
 	selection.reset();
 	address.clear();
 }
@@ -318,7 +319,8 @@ try {
 
 	case PondRequestCommand::FOLLOW:
 		if (!current.MatchId(id) ||
-		    current.command != PondRequestCommand::QUERY)
+		    current.command != PondRequestCommand::QUERY ||
+		    current.continue_)
 			throw SimplePondError{"Misplaced FOLLOW"};
 
 		if (current.follow)
@@ -344,8 +346,8 @@ try {
 		if (!current.filter.sites.empty())
 			throw SimplePondError{"FILTER_SITE and GROUP_SITE are mutually exclusive"};
 
-		if (current.follow)
-			throw SimplePondError{"FOLLOW and GROUP_SITE are mutually exclusive"};
+		if (current.follow || current.continue_)
+			throw SimplePondError{"FOLLOW/CONTINUE and GROUP_SITE are mutually exclusive"};
 
 		if (current.group_site.max_sites > 0)
 			throw SimplePondError{"Duplicate GROUP_SITE"};
@@ -404,8 +406,8 @@ try {
 		    current.command != PondRequestCommand::QUERY)
 			throw SimplePondError{"Misplaced WINDOW"};
 
-		if (current.follow)
-			throw SimplePondError{"FOLLOW and WINDOW are mutually exclusive"};
+		if (current.follow || current.continue_)
+			throw SimplePondError{"FOLLOW/CONTINUE and WINDOW are mutually exclusive"};
 
 		if (current.HasWindow())
 			throw SimplePondError{"Duplicate WINDOW"};
@@ -500,6 +502,27 @@ try {
 		if (!current.filter.duration.HasLonger())
 			throw SimplePondError{"Malformed FILTER_DURATION_LONGER"};
 		return BufferedResult::AGAIN;
+
+	case PondRequestCommand::CONTINUE:
+		if (!current.MatchId(id) ||
+		    current.command != PondRequestCommand::QUERY ||
+		    current.follow)
+			throw SimplePondError{"Misplaced CONTINUE"};
+
+		if (current.continue_)
+			throw SimplePondError{"Duplicate CONTINUE"};
+
+		if (current.HasGroupSite())
+			throw SimplePondError{"CONTINUE and GROUP_SITE are mutually exclusive"};
+
+		if (current.HasWindow())
+			throw SimplePondError{"CONTINUE and WINDOW are mutually exclusive"};
+
+		if (!payload.empty())
+			throw SimplePondError{"Malformed CONTINUE"};
+
+		current.continue_ = true;
+		return BufferedResult::AGAIN;
 	}
 
 	throw SimplePondError{"Command not implemented"};
@@ -676,7 +699,7 @@ Connection::OnBufferedWrite()
 		/* no more sites, end this response */
 	}
 
-	if (current.follow) {
+	if (current.follow || current.continue_) {
 		current.selection->AddAppendListener(*this);
 	} else {
 		Send(current.id, PondResponseCommand::END, {});

src/Connection.hxx

@@ -35,7 +35,7 @@ class Connection final
 
 	struct Request {
 		uint16_t id = 0;
-		bool follow = false;
+		bool follow = false, continue_ = false;
 		PondRequestCommand command = PondRequestCommand::NOP;
 
 		Filter filter;

src/Protocol.hxx

@@ -35,6 +35,8 @@ enum class PondRequestCommand : uint16_t {
 	 * Option for #QUERY which follows incoming new records
 	 * instead of printing past ones.  The response never ends
 	 * until the client sends #CANCEL (or closes the connection).
+	 *
+	 * This is similar to #CONTINUE, but only prints new records.
 	 */
 	FOLLOW = 5,
 
@@ -127,6 +129,16 @@ enum class PondRequestCommand : uint16_t {
 	 * a 64 bit unsigned integer [microseconds].
 	 */
 	FILTER_DURATION_LONGER = 19,
+
+	/**
+	 * Option for #QUERY which follows incoming new records after
+	 * printing past ones.  The response never ends until the
+	 * client sends #CANCEL (or closes the connection).
+	 *
+	 * This is similar to #FOLLOW, but also prints matching past
+	 * records.
+	 */
+	CONTINUE = 20,
 };
 
 enum class PondResponseCommand : uint16_t {

src/client/Main.cxx

@@ -65,7 +65,7 @@ struct QueryOptions {
 
 	bool jsonl = false;
 
-	bool follow = false;
+	bool follow = false, continue_ = false;
 	bool raw = false;
 	bool gzip = false;
 #ifdef HAVE_LIBGEOIP
@@ -202,9 +202,17 @@ ParseFilterItem(Filter &filter, PondGroupSitePayload &group_site,
 		options.per_site_filename = per_site_filename;
 	} else if (StringIsEqual(p, "--per-site-nested")) {
 		options.per_site_nested = true;
-	} else if (StringIsEqual(p, "--follow"))
+	} else if (StringIsEqual(p, "--follow")) {
+		if (options.continue_)
+			throw "Cannot use both --follow and --continue";
+
 		options.follow = true;
-	else if (StringIsEqual(p, "--raw"))
+	} else if (StringIsEqual(p, "--continue")) {
+		if (options.follow)
+			throw "Cannot use both --follow and --continue";
+
+		options.continue_ = true;
+	} else if (StringIsEqual(p, "--raw"))
 		options.raw = true;
 	else if (StringIsEqual(p, "--gzip"))
 		options.gzip = true;
@@ -364,6 +372,9 @@ Query(const PondServerSpecification &server, ConstBuffer<const char *> args)
 	if (options.follow)
 		client.Send(id, PondRequestCommand::FOLLOW);
 
+	if (options.continue_)
+		client.Send(id, PondRequestCommand::CONTINUE);
+
 	client.Send(id, PondRequestCommand::COMMIT);
 
 	struct pollfd pfds[] = {
@@ -588,7 +599,7 @@ try {
 			   "\n"
 			   "Commands:\n"
 			   "  query\n"
-			   "    [--follow]\n"
+			   "    [--follow] [--continue]\n"
 			   "    [--raw] [--gzip]\n"
 #ifdef HAVE_LIBGEOIP
 			   "    [--geoip]\n"