Upload a file in chunks

[azhdari]

Hi,

I try to upload a file to a TUS server, using this example.

private static class App<RT>
        where RT : struct,
        HasCancel<RT>,
        HasConsole<RT>,
        HasFile<RT>,
        HasTextRead<RT>,
        HasHttpClient<RT>
    {
        public static Effect<RT, Unit> MainEffect(Uri url, string path) =>
            File<RT>.openRead(path)
            | Stream<RT>.read(1048576) // 1mb
            | UploadToStorage(url)
            | WaitToComplete;

        static Pipe<RT, SeqLoan<byte>, Aff<RT, object>, Unit> UploadToStorage(Uri url) =>
            from c in awaiting<SeqLoan<byte>>()
            from _ in yield(UploadFilePartial(url,c))
            select unit;

        static Consumer<RT, Aff<RT, object>, Unit> WaitToComplete =>
            from l in awaiting<Aff<RT, object>>()
            from _ in l
            select unit;

        static Aff<RT, object> UploadFilePartial(Uri url, SeqLoan<byte> bytes) {
            var prepareHeaders = fun(
                (HttpClient h) => {
                    h.DefaultRequestHeaders.AddOrUpdate("tus-resumable", "1.0.0");
                    h.DefaultRequestHeaders.AddOrUpdate("upload-offset", startByteIndex.ToString());
                    return h;
                }
            );
            
            return
                from _ in HttpClientEff<RT>.PatchFileBytes(
                    prepareHeaders,
                    url,
                    bytes.ToReadOnlySpan().
                        ToArray()
                )
                select _;
        }
    }

first of all it seems wrong: Pipe<,, Aff<>,>.

Pipe<RT, SeqLoan<byte>, Aff<RT, object>, Unit>

second, is it really working?

static Consumer<RT, Aff<RT, object>, Unit> WaitToComplete =>
     from l in awaiting<Aff<RT, object>>()
     from _ in l // <-- here
     select unit;

how can I get index of the first byte for every chunk?

h.DefaultRequestHeaders.AddOrUpdate("upload-offset", startByteIndex.ToString());

I have to mention, the App<Runtime>.MainEffect() function will be called from somewhere else, and it's not the entry point. So it should work with other Affs.

How can I accomplish this goal?
Thank you

[louthy]

Any reason you're passing the resulting Aff from UploadFilePartial to yield? Seems a bit odd.

Can't you do this?

        static Pipe<RT, SeqLoan<byte>, object, Unit> UploadToStorage(Uri url) =>
            from c in awaiting<SeqLoan<byte>>()
            from r in UploadFilePartial(url, c)
            from _ in yield(r)
            select unit;

It also doesn't seem like the WaitToComplete consumer is necessary. So maybe change UploadToStorage to:

        static Consumer<RT, SeqLoan<byte>, Unit> UploadToStorage(Uri url) =>
            from c in awaiting<SeqLoan<byte>>()
            from r in UploadFilePartial(url,c)
            select unit;

That will close the effect correctly.

how can I get index of the first byte for every chunk?

SeqLoan<A> is just a Seq that will free up its backing array after the pipe has processed it. So, you can use most of the Seq operations on it: i.e. c.HeadOrNone()

[azhdari]

Thank you so much, it's so cleaner now. I'm new at Pipe/Consumer by the way.

you can use most of the Seq operations on it: i.e. c.HeadOrNone()

HeadOrNone() gives us first element in Seq, not the index of it in the origin collection. Even though I cannot find HeadOrNone() for SeqLoan.
SeqLoan has an internal field start, does it hold start index of chunk in origin collection? If so, maybe it's a good idea to make it accessible, or add another field for ChunkNumber.

[azhdari]

I've made some naive changes in Stream.read():

public static class Stream<RT> where RT : struct, HasCancel<RT>
{
    /// <summary>
    /// Get a pipe of chunks from a Stream
    /// </summary>
    public static Pipe<RT, Stream, (SeqLoan<byte> Chunk, ChunkInfo ChunkInfo), Unit> readWithInfo(int chunkSize)
    {
        return from fs in Proxy.awaiting<Stream>()
            from bt in Proxy.enumerate2(chunks(fs, chunkSize))
            from un in Proxy.yield(bt)
            select unit;

        static async IAsyncEnumerable<(SeqLoan<byte>, ChunkInfo)> chunks(Stream fs, int chunkSize)
        {
            var pool = ArrayPool<byte>.Shared;
            var index = 0;
            while (true)
            {
                var buffer = pool.Rent(chunkSize);
                var count  = await fs.ReadAsync(buffer, 0, chunkSize).ConfigureAwait(false);
                if (count < 1)
                {
                    pool.Return(buffer);
                    yield break;
                }

                var info = new ChunkInfo(index, index * chunkSize, count);
                yield return (buffer.ToSeqLoanUnsafe(count, pool), info);

                index += 1;
            }
        }
    }
}

public record ChunkInfo(
    int ChunkIndex,
    int StartItemIndex,
    int Count
);