How to async pipe console stdout to JsonSerializer.DeserializeAsync()?

[ptr727]

I am currently reading a large json doc from stdout to in memory gzip stream, and when cli is done read the gzip memory stream into JsonSerializer.Deserialize().

I am trying to async read from stdin and async read json to avoid memory and latency overhead.

Something like this:

        public async Task<(bool, string, List<FfMpegToolJsonSchema.Packet>)> GetPacketListAsync(
            Command command
        )
        {
            using MemoryStream memoryStream = new();
            StringBuilder stdErrorBuffer = new();
            CommandTask<CommandResult> task = command
                .WithStandardOutputPipe(PipeTarget.ToStream(memoryStream, true))
                .WithStandardErrorPipe(PipeTarget.ToStringBuilder(stdErrorBuffer))
                .WithValidation(CommandResultValidation.None)
                .ExecuteAsync(CancellationToken.None, Program.CancelToken());
   
            CommandResult result = await task;

            FfMpegToolJsonSchema.PacketInfo packetInfo =
                await JsonSerializer.DeserializeAsync<FfMpegToolJsonSchema.PacketInfo>(
                    memoryStream,
                    ConfigFileJsonSchema.JsonReadOptions,
                    Program.CancelToken()
                );

            return (
                task.Task.IsCompletedSuccessfully && result.ExitCode == 0,
                stdErrorBuffer.ToString(),
                packetInfo?.Packets
            );
        }

This fails with an exception where the json deserializer did not find a json token, it looks like the json read does not wait for the stream to have data.

What would a better / correct way be to use cliwrap to async pipe stdout to json?

[Tyrrrz]

You'll have to create your own PipeTarget. Rough sketch:

FfMpegToolJsonSchema.PacketInfo packetInfo;

var stdOutTarget = PipeTarget.Create(async (stdOutStream, cancellationToken) =>
{
    packetInfo = await JsonSerializer.DeserializeAsync<FfMpegToolJsonSchema.PacketInfo>(
        stdOutStream,
        ConfigFileJsonSchema.JsonReadOptions,
        cancellationToken
    );
});

var cmd = command
  .WithStandardOutputPipe(stdOutTarget)
  .WithStandardErrorPipe(PipeTarget.ToStringBuilder(stdErrorBuffer))
  .WithValidation(CommandResultValidation.None);

You can also just derive from PipeTaget and override CopyAsync(...) if you'd rather define your own type instead of using a one-off anonymous instance.

You can refactor it to your liking, but that's the idea.

[ptr727]

Ok, thx. I was wondering if a pull model would be better to read packet by packet. In some scenarios that is even more optimal for e.g. scenarios where I only need aggregate values.

[Tyrrrz]

The built-in pull model splits entries by line breaks. When consuming JSON it may be suboptimal since it legally might not have any line breaks altogether.

[ptr727]

Hmm, yes, that would be problematic for JSON being streamed as a string with no line breaks.

Does the line-break logic apply to the push model as well?

Else I can experiment with DeserializeAsyncEnumerable or Utf8JsonStreamReader wait for the array to start then async read array elements from the stream one-by-one.

[Tyrrrz]

Does the line-break logic apply to the push model as well?

Yes.

Ultimately, all of the execution models are based on the piping building blocks, they're just provided to simplify very common scenarios. Your scenario is not very common, hence why you'd be using low-level piping building blocks instead. But you can wrap it all in an extension method to make it cleaner.

The code I showed above is the most optimal solution for your problem, from a technical standpoint.