It’s been a bit since my last post, and it was mostly me trying to figure out what to talk about next. I thought this would be an interesting post. NOTE: the code for this blog can be found here.
So, uploading large files directly from a web site can be a burden. If the client uploads the entire file at once then the web site will freeze. That’s where Flow.js comes into the picture.
Flow.js handles uploading large files by splitting them up into smaller pieces and uploading each piece to a server. The server can then assemble the pieces back into the original file.
In this blog, I will not be accessing Flow.js directly, I’ll be using a package for Angular called Ngx-Flow. This package wraps the Flow.js functionality into a nice directive. Let’s test it out first. I must add that I had to do some modifications to the Angular project to get it to work on Angular 14. Just check my package.json for extra packages (there’s only one).
Grab the code here and open the API project (from the API folder) in VS2022 (or your IDE of choice) and run it. Next navigate the the web folder and start Angular by typing in ‘ng serve’ (NOTE: this uses Angular 14).
There are two methods displayed (simple and advanced). The advanced simply displays more details about the files and allows for auto uploading. Either drag some files to either of the drag and drop area(s) or select the ‘choose files’ button. After selecting the files, hit the ‘start upload’ button and they’ll be transferred to the API and reassembled. The API stores the files in the ‘Uploads’ folder. You should see any transferred files there.
Let’s look at the front-end code first. I’ll go over the ‘advanced’ example (which I took from the ngx-flow example, here, and modified for Angular 14):
<h2>Advanced example</h2>
<p>Is flowjs supported by the browser? {{flow.flowJs?.support ? "yes" : "no"}}</p>
<div>
<input type="checkbox" [(ngModel)]="autoupload"> Autoupload
</div>
<ng-container #flowAdvanced="flow" [flowConfig]="{target: 'https://localhost:7015/upload'}"></ng-container>
<input type="file"
flowButton
[flow]="flowAdvanced.flowJs"
[flowAttributes]="{accept: 'image/*, video/*'}">
<div class="drop-area"
flowDrop
[flow]="flowAdvanced.flowJs">
<span>Drop files here</span>
</div>
<div class="transfers">
<div class="transfer"
[ngClass]="{'transfer--error': transfer.error, 'transfer--success': transfer.success}"
*ngFor="let transfer of (flowAdvanced.transfers$ | async)?.transfers; trackBy: trackTransfer">
<div class="name">name: {{transfer.name}}</div>
<div>progress: {{transfer.progress | percent}}</div>
<div>size: {{transfer.size | number: '1.0'}} bytes</div>
<div>current speed: {{transfer.currentSpeed | number: '1.0'}} bytes/s</div>
<div>average speed: {{transfer.averageSpeed | number: '1.0'}} bytes/s</div>
<div>time ramining: {{transfer.timeRemaining}}s</div>
<div>paused: {{transfer.paused}}</div>
<div>success: {{transfer.success}}</div>
<div>complete: {{transfer.complete}}</div>
<div>error: {{transfer.error}}</div>
<!-- <div>
<img [flowSrc]="transfer">
</div> -->
<div>
<button (click)="flowAdvanced.pauseFile(transfer)">pause</button>
<button (click)="flowAdvanced.resumeFile(transfer)">resume</button>
<button (click)="flowAdvanced.cancelFile(transfer)">cancel</button>
</div>
</div>
</div>
<button type="button" (click)="flowAdvanced.upload()" [disabled]="!(flowAdvanced.somethingToUpload$ | async)">Start upload</button>
<button type="button" (click)="flowAdvanced.cancel()" [disabled]="!(flowAdvanced.transfers$ | async)?.transfers?.length">Cancel all</button>
Total progress: {{(flowAdvanced.transfers$ | async)?.totalProgress | percent}}
This is the html file and it mostly just uses the ngx-flow directive. You can see that the code checks for flow support and then sets the directive (#flowAdvanced) and sets the flow config to upload to the API’s upload endpoint.
Then, it provides a file open input and a drag-and-drop area. After that, it iterates through the transfers, setting various properties to be displayed (for example progress and file size).
For each selected file, you can pause, resume, or cancel the upload. After that, there’s the Upload button and the total progress field. Let’s look at the typescript file next (app.component.ts):
import { Component, ViewChild } from '@angular/core';
import { FlowDirective, Transfer } from '@flowjs/ngx-flow';
import { Subscription } from 'rxjs';
@Component({
selector: 'app-root',
templateUrl: './app.component.html',
styleUrls: ['./app.component.scss']
})
export class AppComponent {
title = 'web';
@ViewChild('flowAdvanced')
public flow!: FlowDirective;
public autoupload: boolean = false;
public autoUploadSubscription!: Subscription;
ngAfterViewInit() {
this.autoUploadSubscription = this.flow.events$.subscribe(event => {
// to get rid of incorrect `event.type` type you need Typescript 2.8+
if (this.autoupload && event.type === 'filesSubmitted') {
this.flow.upload();
}
});
}
ngOnDestroy() {
this.autoUploadSubscription?.unsubscribe();
}
trackTransfer(idx: number, transfer?: Transfer) {
return transfer?.id ?? 0;
}
}
This doesn’t do too much. Inside the AppComponent, it sets the FlowDirective from the html file (ViewChild). In the ngAftearViewInit function, it sets up an event subscription, and if autoupload is set to true, then it uploads the file(s) immediately.
In ngOnDestroy, it unsubscribes from the event and trackTransfer is simply used to track items (from the ngFor in the html) by the transfer id. Let’s look at the API.
The API has one controller that is used for handling the uploads and it has a few helper files which assist in saving the files. NOTE: I used some code from this Github repo and modified it for .NET 6.
The upload controller gets injected with the IFlowUploadProcessorCore which is the main entry for saving the files (and is set to scoped lifetime). It contains two endpoints. The first one does the actual uploading. It gets the FlowMetaData object from the URI and calls ‘ProcessUploadChunkRequest’ with the request, the meta-data, and the upload path. The second one is used by flow to verify a chunk of the file was received or not. Let’s look at the FlowMetaData object:
public class FlowMetaData
{
public long FlowChunkNumber { get; set; }
public long FlowChunkSize { get; set; }
public long FlowCurrentChunkSize { get; set; }
public long FlowTotalSize { get; set; }
public string FlowIdentifier { get; set; }
public string FlowFilename { get; set; }
public string FlowRelativePath { get; set; }
public int FlowTotalChunks { get; set; }
public long FileOffset
{
get { return FlowChunkSize * (FlowChunkNumber - 1); }
}
}This object gets set by flow and is used to save the file. I’ll talk about each field individually:
FlowChunkNumber: this is the current number of the file piece. Note, it starts at 1.
FlowChunkSize: size (in bytes) of the chunks that flow is sending.
FlowCurrentChunkSize: size (in bytes) of this chunk. (This can be different than FlowChunkSize if the file is either smaller than the normal chunk size or we’re on the last chunk).
FlowTotalSize: the total size of the file.
FlowIdentifier: this is a unique identifier for the uploaded file.
FlowFilename: this is the original filename.
FlowRelativePath: the path of the file(s) uploaded. this is used if the user drops a directory into the drag-and-drop area.
FlowTotalChunks: the total number of pieces that this file has been split up into.
FileOffset: this is used when writing the chunk to the file.
Let’s look at FlowUploadProcessorCore next:
public interface IFlowUploadProcessorCore
{
DateTime CompletedDateTime { get; }
bool IsComplete { get; }
bool HasRecievedChunk(FlowMetaData chunkMeta);
Task<bool> ProcessUploadChunkRequest(HttpRequest request, FlowMetaData filePart, string uploadPath);
}
public class FlowUploadProcessorCore : IFlowUploadProcessorCore
{
private static readonly object lockObject = new object();
public FlowUploadProcessorCore(IMemoryCache memoryCache)
{
this.memoryCache = memoryCache;
}
//================================================================================
// Class Methods
//================================================================================
#region Methods
/// <summary>
/// Track our in progress uploads, by using a cache, we make sure we don't accumulate memory
/// </summary>
private readonly IMemoryCache memoryCache;
private FileMetaData? GetFileMetaData(string flowIdentifier)
{
if (memoryCache.TryGetValue(flowIdentifier, out var fileMetaData))
{
return fileMetaData as FileMetaData;
}
return null;
}
/// <summary>
/// Keep an upload in cache for two hours after it is last used
/// </summary>
private static MemoryCacheEntryOptions DefaultCacheItemPolicy()
{
return new MemoryCacheEntryOptions()
{
SlidingExpiration = TimeSpan.FromMinutes(5)
};
}
/// <summary>
/// (Thread Safe) Marks a chunk as recieved.
/// </summary>
private bool RegisterSuccessfulChunk(FlowMetaData chunkMeta)
{
var fileMeta = GetFileMetaData(chunkMeta.FlowIdentifier);
if (fileMeta == null)
{
lock (lockObject)
{
fileMeta = GetFileMetaData(chunkMeta.FlowIdentifier);
if (fileMeta == null)
{
fileMeta = new FileMetaData(chunkMeta);
memoryCache.Set(chunkMeta.FlowIdentifier, fileMeta, DefaultCacheItemPolicy());
}
}
}
fileMeta.RegisterChunkAsReceived(chunkMeta);
lock (lockObject)
{
memoryCache.Set(chunkMeta.FlowIdentifier, fileMeta, DefaultCacheItemPolicy());
}
if (fileMeta.IsComplete)
{
// Since we are using a cache and memory is automatically disposed,
// we don't need to do this, so we won't so we can keep a record of
// our completed uploads.
memoryCache.Remove(chunkMeta.FlowIdentifier);
}
return fileMeta.IsComplete;
}
public bool HasRecievedChunk(FlowMetaData chunkMeta)
{
var fileMeta = GetFileMetaData(chunkMeta.FlowIdentifier);
bool wasRecieved = fileMeta != null && fileMeta.HasChunk(chunkMeta);
return wasRecieved;
}
public bool IsComplete { get; private set; }
public DateTime CompletedDateTime { get; private set; }
public async Task<bool> ProcessUploadChunkRequest(HttpRequest request, FlowMetaData filePart, string uploadPath)
{
var fileData = request.Form.Files[0];
var filePath = $"{uploadPath}/{filePart.FlowFilename}";
using (var flowFileStream = File.Open(filePath, FileMode.OpenOrCreate, FileAccess.Write, FileShare.Write))
{
flowFileStream.SetLength(filePart.FlowTotalSize);
flowFileStream.Seek(filePart.FileOffset, 0);
{
await fileData.CopyToAsync(flowFileStream);
}
}
IsComplete = RegisterSuccessfulChunk(filePart);
if (IsComplete)
{
CompletedDateTime = DateTime.Now;
}
return IsComplete;
}
#endregion
}
First thing, is IMemoryCache is injected into the constructor. This is used to keep track of the file upload progress.
GetFileMetaData: This tries to retrieve the file data from the cache. If it doesn’t exist then it returns null.
DefaultCacheItemPolicy: This is used to set the cache retention policy. Currently, it’s set to a sliding expiration for 5 minutes which means the cache item will get removed 5 minutes from the last time it was accesssed.
RegisterSuccessfulChunk: This function takes the FlowMetaData object and retrieves the FileMetaData from the cache. If it doesn’t exist, then it creates a new one. Next, it registers the chunk as received and pushes it back into the cache. If the upload is complete, then it removes the item from the cache.
HasRecievedChunk: This function gets the FileMetaData from the cache and returns if the chunk was received (from FileMetaData).
ProcessUploadChunkRequest: This function is called from the REST upload endpoint and does the actual saving of the file parts. It grabs the file data from the request, sets the path, and then opens the file for writing. Then, it sets the file length and seeks to the offset for this chunk then copies the data to the open file. (One thing to note is that the created stream allows multiple writes at the same time which I did not know was supported.) After this, it registers the chunk as complete and returns if the file has completed.
The FileMetaData is the object that is cached. It tracks the file completion progress:
/// <summary>
/// Our own internal metadata to track the progress of a download.
/// </summary>
class FileMetaData
{
private static long ChunkIndex(long chunkNumber)
{
return chunkNumber - 1;
}
public FileMetaData(FlowMetaData flowMeta)
{
FlowMeta = flowMeta;
ChunkArray = new bool[flowMeta.FlowTotalChunks];
TotalChunksReceived = 0;
}
public bool[] ChunkArray { get; set; }
/// <summary>
/// Chunks can come out of order, so we must track how many chunks
/// we have successfully recieved to determine if the upload is complete.
/// </summary>
public int TotalChunksReceived { get; set; }
public FlowMetaData FlowMeta { get; set; }
public bool IsComplete
{
get
{
return (TotalChunksReceived == FlowMeta.FlowTotalChunks);
}
}
public void RegisterChunkAsReceived(FlowMetaData flowMeta)
{
ChunkArray[ChunkIndex(flowMeta.FlowChunkNumber)] = true;
TotalChunksReceived++;
}
public bool HasChunk(FlowMetaData flowMeta)
{
return ChunkArray[ChunkIndex(flowMeta.FlowChunkNumber)];
}
}
The constructor requires a FlowMetaData object, and it sets a new array of boolean to the length of total chunks. This allows it to track which chunks are complete.
IsComplete: this returns if the all of the chunks have been received and the upload is compelte.
RegisterChunkAsReceived: This function sets the Chunk Array (booleans) for the chunk number that was received and increments TotalChunksReceived.
HasChunk: This simply returns the boolean value from the chunk array, using the chunk number from the FlowMetaData object. (Ultimately, this is used by the GET endpoint.)
I’ve verified this with a few 8gig files and it works! Thanks for stopping by!