2
\$\begingroup\$

I am coding a csv file filtering api in ASP.NET Core, which takes a csv file located at a given uri (and retrieved via the query string) convert the content to a collection of records, then apply a business rule to mark some records as pass or fail and then convert them as json/xml as part of another, output stream (the response returned by the web api).

The main constraint is that the file can big kinda huge and I am trying to leverage async/await as much as I can to prevent the controller action to get stuck for too long. Also I am trying to use exclusively Streams in order to prevent peaks of memory from happening.

I am seeking some help for improving my code about

  • Naming (I am still not sure that what I am doing can be qualified as "filtering" but I could not come up with a better name).
  • OO Design, I am looking for a better way to separate my business validation from the streaming process and exception handling (ie. a row cannot be deserialized as CsvRecord)
  • Maintainability, resilient to requirements changes
  • Error / Exception Handling Improvement not sure how to improve my controller action if say the csvUri leads to nowhere or if the headers is not validated, catching the exception, using Either?

Note: I am using 2 libraries:

I also struggled to have used FluentValidation to externalize the business validation

The Program and Startup with the dependency injection initialization:

public class Program
{
    public static void Main(string[] args)
    {
        CreateWebHostBuilder(args).Build().Run();
    }

    public static IWebHostBuilder CreateWebHostBuilder(string[] args) =>
        WebHost.CreateDefaultBuilder(args)
            .UseStartup<Startup>();
}

public class Startup
{
    public Startup(IConfiguration configuration)
    {
        Configuration = configuration;
    }

    public IConfiguration Configuration { get; }

    // This method gets called by the runtime. Use this method to add services to the container.
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddTransient<IAsyncStreamFilterFactory, MyAsyncStreamFilterFactory>();
        services.AddMvc().SetCompatibilityVersion(CompatibilityVersion.Version_2_2);
    }

    // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
    public void Configure(IApplicationBuilder app, IHostingEnvironment env)
    {
        if (env.IsDevelopment())
        {
            app.UseDeveloperExceptionPage();
        }
        else
        {
            // The default HSTS value is 30 days. You may want to change this for production scenarios, see https://aka.ms/aspnetcore-hsts.
            app.UseHsts();
        }

        app.UseHttpsRedirection();
        app.UseMvc();
    }
}

The controller and its action taking the csv uri, extracting its content to related records, filter some according to some business rules and serialize back what have been filtered to the response body stream.

[Route("[controller]")]
[ApiController]
public class FilterController : ControllerBase
{
    private readonly IAsyncStreamFilterFactory _asyncStreamFilterFactory;

    public FilterController(IAsyncStreamFilterFactory asyncStreamFilterFactory)
    {
        _asyncStreamFilterFactory = asyncStreamFilterFactory;
    }

    // GET / POST filter?csvUri=...
    [HttpGet]
    [HttpPost]
    [Produces("application/json", "application/xml")]
    public async Task FilterAsync([FromQuery(Name = "csvUri")] string csvUri)
    {
        var acceptHeader = Request.Headers["Accept"];
        switch (acceptHeader)
        {
            case "application/json":
                var jsonAsyncStreamFilter = _asyncStreamFilterFactory.Create(AsyncStreamFilterFormat.Json);
                await jsonAsyncStreamFilter.FilterAsyncFromUriAsync(csvUri, Response.Body);
                return;

            case "application/xml":
                var xmlAsyncStreamFilter = _asyncStreamFilterFactory.Create(AsyncStreamFilterFormat.Xml);
                await xmlAsyncStreamFilter.FilterAsyncFromUriAsync(csvUri, Response.Body);
                return;

            default:
                throw new UnsupportedMediaTypeException($"{acceptHeader} is not a supported format", MediaTypeHeaderValue.Parse(acceptHeader));
        }
    }
}

The IAsyncStreamFilterFactory and IAsyncStreamFilter implementations:

public enum AsyncStreamFilterFormat
{
    Json,
    Xml
}

public interface IAsyncStreamFilterFactory
{
    IAsyncStreamFilter Create(AsyncStreamFilterFormat format);
}

public class MyAsyncStreamFilterFactory : IAsyncStreamFilterFactory
{
    public IAsyncStreamFilter Create(AsyncStreamFilterFormat format)
    {
        switch (format)
        {
            case AsyncStreamFilterFormat.Json:
                return new MyAsyncCsvStreamFilter(new MyJsonAsyncEnumerableToStreamSerializer());
            case AsyncStreamFilterFormat.Xml:
                return new MyAsyncCsvStreamFilter(new MyXmlAsyncEnumerableToStreamSerializer());
            default:
                throw new InvalidEnumArgumentException(nameof(format), (int)format, typeof(AsyncStreamFilterFormat));
        }
    }
}

public interface IAsyncStreamFilter
{
    Task FilterAsync(Stream input, Stream output);
}

public class MyAsyncCsvStreamFilter : IAsyncStreamFilter
{
    private readonly IAsyncConverterIn<IAsyncEnumerable<Result>, Stream> _asyncEnumerableToStreamSerializer;

    public MyAsyncCsvStreamFilter(
        IAsyncConverterIn<IAsyncEnumerable<Result>, Stream> asyncEnumerableToStreamSerializer)
    {
        _asyncEnumerableToStreamSerializer = asyncEnumerableToStreamSerializer;
    }

    private static bool IsRecordBusinessValid(CsvRecord record) => 
        record.C + record.D > 100;

    private static Result CreateOkResult(CsvRecord record, long lineNumber) =>
        new ErrorResult
        {
            LineNumber = lineNumber,
            ErrorMessage = "The Sum of C and D is lesser than 100..."
        };

    private static Result CreateExceptionErrorResult(Exception exception, long lineNumber) =>
        new ErrorResult
        {
            LineNumber = lineNumber,
            ErrorMessage = exception.Message
        };

    private Result CreateBusinessErrorResult(CsvRecord record, long lineNumber) =>
        new OkResult
        {
            LineNumber = lineNumber,
            ConcatAB = record.A + record.B,
            SumCD = record.C + record.D
        };

    public async Task FilterAsync(Stream input, Stream output)
    {
        var records = input.AsCsvReader().EnumerateTryRecordsAsync<CsvRecord>();

        var results = records.Select((record, index) =>
        {
            var lineNumber = index + 1;

            return record.IsFaulted
                ? CreateExceptionErrorResult(record.Exception, lineNumber)
                : IsRecordBusinessValid(record.Value)
                    ? CreateOkResult(record.Value, lineNumber)
                    : CreateBusinessErrorResult(record.Value, lineNumber);
        });

        await _asyncEnumerableToStreamSerializer.ConvertAsync(results, output);
    }
}


public interface IAsyncConverterIn<in TInput, in TOutput>
{
    Task ConvertAsync(TInput input, TOutput output);
}

public class MyJsonAsyncEnumerableToStreamSerializer : IAsyncConverterIn<IAsyncEnumerable<Result>, Stream>
{
    public async Task ConvertAsync(IAsyncEnumerable<Result> input, Stream output)
    {
        using (var streamWriter = output.AsStreamWriter())
        {
            await streamWriter.WriteLineAsync("[");
            await input.Detailed().ForEachAsync(async item =>
                {
                    await streamWriter.WriteLineAsync("  " + item.Value.ToJson() + (!item.IsLast ? string.Empty : ","));
                }
            );
            await streamWriter.WriteLineAsync("]");
        }
    }
}

public class MyXmlAsyncEnumerableToStreamSerializer : IAsyncConverterIn<IAsyncEnumerable<Result>, Stream>
{
    public async Task ConvertAsync(IAsyncEnumerable<Result> input, Stream output)
    {
        using (var streamWriter = output.AsStreamWriter())
        {
            await streamWriter.WriteLineAsync("<Array>");
            await input.ForEachAsync(async item =>
                {
                    await streamWriter.WriteLineAsync("  " + item.ToXml());
                }
            );
            await streamWriter.WriteLineAsync("</Array>");
        }
    }
}

public class CsvRecord
{
    [Name("columnA")]
    public string A { get; set; }

    [Name("columnB")]
    public string B { get; set; }

    [Name("columnC")]
    public int C { get; set; }

    [Name("columnD")]
    public int D { get; set; }
}

public abstract class Result
{
    public long LineNumber { get; set; }

    public abstract string Type { get; }
}

public class OkResult : Result
{
    public override string Type { get; } = "ok";

    public string ConcatAB { get; set; }
    public int SumCD { get; set; }
}

public class ErrorResult : Result
{
    public override string Type { get; } = "error";

    public string ErrorMessage { get; set; }
}

The extension Methods for Streams, CsvReader,IAsyncEnumerable<T> and IAsyncStreamFilter:

public static class StreamExtensions
{
    public static StreamReader AsStreamReader(this Stream stream) =>
        new StreamReader(stream);

    public static StreamWriter AsStreamWriter(this Stream stream) =>
        new StreamWriter(stream);

    public static CsvReader AsCsvReader(this Stream stream) =>
        stream.AsStreamReader().AsCsvReader();
}

public static class StreamReaderExtensions
{
    public static CsvReader AsCsvReader(this StreamReader streamReader) =>
        new CsvReader(streamReader);
}

public static class CsvReaderExtensions
{
    public static async Task<bool> ReadAndValidateHeaderAsync<TRecord>(this CsvReader csvReader, bool detectColumnCountChanges = false)
    {
        csvReader.Configuration.DetectColumnCountChanges = detectColumnCountChanges;
        var hasReadHeader = await csvReader.ReadAsync() && !csvReader.ReadHeader();

        csvReader.ValidateHeader<TRecord>();

        return hasReadHeader;
    }

    public static IAsyncEnumerable<TRecord> EnumerateRecordsAsync<TRecord>(this CsvReader csvReader, bool mustHaveHeader = true) => 
    return new AsyncEnumerable<TRecord>(async yield =>
    {
        if (mustHaveHeader)
        {
            await csvReader.ReadAndValidateHeaderAsync<TRecord>();
        }

        while (await csvReader.ReadAsync())
        {
            await yield.ReturnAsync(csvReader.GetRecord<TRecord>());
        }
    });


    public static IAsyncEnumerable<TryResult<TRecord>> EnumerateTryRecordsAsync<TRecord>(this CsvReader csvReader, bool mustHaveHeader = true) => 
    return new AsyncEnumerable<TryResult<TRecord>>(async yield =>
    {
        if (mustHaveHeader)
        {
            await csvReader.ReadAndValidateHeaderAsync<TRecord>();
        }

        csvReader.ValidateHeader<TRecord>();

        while (await csvReader.ReadAsync())
        {
            Try<TRecord> tryRecord = () => csvReader.GetRecord<TRecord>();
            await yield.ReturnAsync(tryRecord.Try());
        }
    });
}

public static class AsyncEnumerableExtensions
{
    public static IAsyncEnumerable<Iteration<TSource>> Detailed<TSource>(this IAsyncEnumerable<TSource> source) =>
        new AsyncEnumerable<Iteration<TSource>>(async yield =>
        {
            using (var enumerator = await source.GetAsyncEnumeratorAsync())
            {
                var isFirst = true;
                var hasNext = await enumerator.MoveNextAsync();
                var index = 0L;
                while (hasNext)
                {
                    var current = enumerator.Current;
                    hasNext = await enumerator.MoveNextAsync();
                    await yield.ReturnAsync(new Iteration<TSource>(index, current, isFirst, !hasNext));
                    isFirst = false;
                    index++;
                }
            }
        });
}


public readonly struct Iteration<T>
{
    public readonly long Index;
    public readonly bool IsFirst;
    public readonly bool IsLast;
    public readonly T Value;

    public Iteration(long index, T value, bool isFirst, bool isLast)
    {
        Index = index;
        IsFirst = isFirst;
        IsLast = isLast;
        Value = value;
    }
}

public static class AsyncStreamFilterExtensions
{
    public static async Task FilterAsyncFromPathAsync(this IAsyncStreamFilter asyncStreamFilter, string path, Stream output)
    {
        var input = File.OpenRead(path);
        await asyncStreamFilter.FilterAsync(input, output);
    }

    public static async Task FilterAsyncFromUriAsync(this IAsyncStreamFilter asyncStreamFilter, string uri, Stream output)
    {
        var input = await new HttpClient().GetStreamAsync(uri);
        await asyncStreamFilter.FilterAsync(input, output);
    }
}
\$\endgroup\$
  • \$\begingroup\$ Try? TryResult? \$\endgroup\$ – Nkosi Apr 14 at 2:56
  • \$\begingroup\$ @Nkosi I am gonna update the post, but Try and TryResult come from github.com/louthy/csharp-monad \$\endgroup\$ – Ehouarn Perret Apr 14 at 9:47
  • 1
    \$\begingroup\$ You should never arbitrarily create new instances of HttpClient, otherwise you will run into issues. It is specifically designed to be reusable/thread-safe. \$\endgroup\$ – Brad M Apr 16 at 13:05
  • \$\begingroup\$ @BradM fair enough, docs.microsoft.com/en-us/dotnet/standard/… \$\endgroup\$ – Ehouarn Perret Apr 16 at 13:37

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.