varlink: Add new varlink_collect method

varlink_collect is meant to be used when the client is willing to wait for the reply from the varlink method, much like varlink_call.
However, unlike varlink_call, it allows the client to collect all "more" replies that may be sent by method before the "final" reply is enqueued.
It aggregates all of these replies into a json variant array that it returns to the client.

src/shared/varlink.c

@@ -240,6 +240,12 @@ struct VarlinkServer {
         bool exit_on_idle;
 };
 
+typedef struct VarlinkCollectContext {
+        JsonVariant *parameters;
+        const char *error_id;
+        VarlinkReplyFlags flags;
+} VarlinkCollectContext ;
+
 static const char* const varlink_state_table[_VARLINK_STATE_MAX] = {
         [VARLINK_IDLE_CLIENT]              = "idle-client",
         [VARLINK_AWAITING_REPLY]           = "awaiting-reply",
@@ -2088,6 +2094,144 @@ int varlink_callb(
         return varlink_call(v, method, parameters, ret_parameters, ret_error_id, ret_flags);
 }
 
+static void varlink_collect_context_free(VarlinkCollectContext *cc) {
+        assert(cc);
+
+        json_variant_unref(cc->parameters);
+        free((char *)cc->error_id);
+}
+
+static int collect_callback(
+                Varlink *v,
+                JsonVariant *parameters,
+                const char *error_id,
+                VarlinkReplyFlags flags,
+                void *userdata) {
+
+        VarlinkCollectContext *context = ASSERT_PTR(userdata);
+        int r;
+
+        assert(v);
+
+        context->flags = flags;
+        /* If we hit an error, we will drop all collected replies and just return the error_id and flags in varlink_collect() */
+        if (error_id) {
+                context->error_id = error_id;
+                return 0;
+        }
+
+        r = json_variant_append_array(&context->parameters, parameters);
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to append JSON object to array: %m");
+
+        return 1;
+}
+
+int varlink_collect(
+                Varlink *v,
+                const char *method,
+                JsonVariant *parameters,
+                JsonVariant **ret_parameters,
+                const char **ret_error_id,
+                VarlinkReplyFlags *ret_flags) {
+
+        _cleanup_(varlink_collect_context_free) VarlinkCollectContext context = {};
+        int r;
+
+        assert_return(v, -EINVAL);
+        assert_return(method, -EINVAL);
+
+        if (v->state == VARLINK_DISCONNECTED)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
+        if (v->state != VARLINK_IDLE_CLIENT)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
+
+        assert(v->n_pending == 0); /* n_pending can't be > 0 if we are in VARLINK_IDLE_CLIENT state */
+
+        /* If there was still a reply pinned from a previous call, now it's the time to get rid of it, so
+         * that we can assign a new reply shortly. */
+        varlink_clear_current(v);
+
+        r = varlink_bind_reply(v, collect_callback);
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to bind collect callback");
+
+        varlink_set_userdata(v, &context);
+        r = varlink_observe(v, method, parameters);
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to collect varlink method: %m");
+
+        while (v->state == VARLINK_AWAITING_REPLY_MORE) {
+
+                r = varlink_process(v);
+                if (r < 0)
+                        return r;
+
+                /* If we get an error from any of the replies, return immediately with just the error_id and flags*/
+                if (context.error_id) {
+                        if (ret_error_id)
+                                *ret_error_id = TAKE_PTR(context.error_id);
+                        if (ret_flags)
+                                *ret_flags = context.flags;
+                        return 0;
+                }
+
+                if (r > 0)
+                        continue;
+
+                r = varlink_wait(v, USEC_INFINITY);
+                if (r < 0)
+                        return r;
+        }
+
+        switch (v->state) {
+
+        case VARLINK_IDLE_CLIENT:
+                break;
+
+        case VARLINK_PENDING_DISCONNECT:
+        case VARLINK_DISCONNECTED:
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(ECONNRESET), "Connection was closed.");
+
+        case VARLINK_PENDING_TIMEOUT:
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(ETIME), "Connection timed out.");
+
+        default:
+                assert_not_reached();
+        }
+
+        if (ret_parameters)
+                *ret_parameters = TAKE_PTR(context.parameters);
+        if (ret_error_id)
+                *ret_error_id = TAKE_PTR(context.error_id);
+        if (ret_flags)
+                *ret_flags = context.flags;
+        return 1;
+}
+
+int varlink_collectb(
+                Varlink *v,
+                const char *method,
+                JsonVariant **ret_parameters,
+                const char **ret_error_id,
+                VarlinkReplyFlags *ret_flags, ...) {
+
+        _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL;
+        va_list ap;
+        int r;
+
+        assert_return(v, -EINVAL);
+
+        va_start(ap, ret_flags);
+        r = json_buildv(&parameters, ap);
+        va_end(ap);
+
+        if (r < 0)
+                return varlink_log_errno(v, r, "Failed to build json message: %m");
+
+        return varlink_collect(v, method, parameters, ret_parameters, ret_error_id, ret_flags);
+}
+
 int varlink_reply(Varlink *v, JsonVariant *parameters) {
         _cleanup_(json_variant_unrefp) JsonVariant *m = NULL;
         int r;

src/shared/varlink.h

@@ -89,6 +89,10 @@ int varlink_sendb(Varlink *v, const char *method, ...);
 int varlink_call(Varlink *v, const char *method, JsonVariant *parameters, JsonVariant **ret_parameters, const char **ret_error_id, VarlinkReplyFlags *ret_flags);
 int varlink_callb(Varlink *v, const char *method, JsonVariant **ret_parameters, const char **ret_error_id, VarlinkReplyFlags *ret_flags, ...);
 
+/* Send method call and begin collecting all 'more' replies into an array, finishing when a final reply is sent */
+int varlink_collect(Varlink *v, const char *method, JsonVariant *parameters, JsonVariant **ret_parameters, const char **ret_error_id, VarlinkReplyFlags *ret_flags);
+int varlink_collectb(Varlink *v, const char *method, JsonVariant **ret_parameters, const char **ret_error_id, VarlinkReplyFlags *ret_flags, ...);
+
 /* Enqueue method call, expect a reply, which is eventually delivered to the reply callback */
 int varlink_invoke(Varlink *v, const char *method, JsonVariant *parameters);
 int varlink_invokeb(Varlink *v, const char *method, ...);