Skip to content

Add CountStream to get TotalCount from non-seekable sources #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vladimir-buravlev merged 1 commit into from
Jul 17, 2025
Merged

Add CountStream to get TotalCount from non-seekable sources #37

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 66 additions & 32 deletions src/VirtoCommerce.ImportModule.CsvHelper/CsvDataReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,23 +14,24 @@ namespace VirtoCommerce.ImportModule.CsvHelper
{
public class CsvDataReader<TCsvImportable, TCsvClassMap> : IImportDataReader where TCsvClassMap : ClassMap
{
private readonly Stream _stream;

private readonly int _pageSize;
private readonly bool _needReadRaw;
private int? _totalCount;
private string _headerRaw;
private bool _disposed;
protected CsvConfiguration CsvConfiguration { get; set; }
protected readonly Stream Stream;
protected readonly Stream CountStream;
protected readonly CsvReader CsvReader;
protected string HeaderRaw;
protected int? TotalCount;

public bool HasMoreResults { get; private set; } = true;

public CsvDataReader(Stream stream, ImportContext context, bool needReadRaw = false)
{
CsvConfiguration = GetConfiguration(context);

_stream = stream;
CsvReader = new CsvReader(new StreamReader(_stream), CsvConfiguration);
Stream = stream;
CsvReader = new CsvReader(new StreamReader(Stream), CsvConfiguration);
CsvReader.Context.RegisterClassMap<TCsvClassMap>();

_pageSize = Convert.ToInt32(context.ImportProfile.Settings.FirstOrDefault(x => x.Name == CsvSettings.PageSize.Name)?.Value ?? 50);
Expand All @@ -41,49 +42,73 @@ public CsvDataReader(Stream stream, ImportContext context, CsvConfiguration csvC
{
CsvConfiguration = MergeWithDefaultConfig(csvConfiguration, context);

_stream = stream;
CsvReader = new CsvReader(new StreamReader(_stream), CsvConfiguration);
Stream = stream;
CsvReader = new CsvReader(new StreamReader(Stream), CsvConfiguration);
CsvReader.Context.RegisterClassMap<TCsvClassMap>();

_pageSize = Convert.ToInt32(context.ImportProfile.Settings.FirstOrDefault(x => x.Name == CsvSettings.PageSize.Name)?.Value ?? 50);
_needReadRaw = needReadRaw;
}

public CsvDataReader(Stream stream, Stream countStream, ImportContext context, bool needReadRaw = false)
: this(stream, context, needReadRaw)
{
CountStream = countStream;
}

public CsvDataReader(Stream stream, Stream countStream, ImportContext context, CsvConfiguration csvConfiguration, bool needReadRaw = false)
: this(stream, context, csvConfiguration, needReadRaw)
{
CountStream = countStream;
}

public virtual async Task<int> GetTotalCountAsync(ImportContext context)
{
if (_totalCount.HasValue)
if (TotalCount.HasValue)
{
return _totalCount.Value;
return TotalCount.Value;
}

Stream stream;
bool leaveOpen;
if (Stream.CanSeek)
{
stream = Stream;
leaveOpen = true;
}
else
{
stream = CountStream ?? throw new InvalidOperationException("Count stream is not provided.");
leaveOpen = false;
}

var streamPosition = 0L;
if (_stream.CanSeek)
if (stream.CanSeek)
{
streamPosition = _stream.Position;
_stream.Seek(0, SeekOrigin.Begin);
streamPosition = stream.Position;
stream.Seek(0, SeekOrigin.Begin);
}

var streamReader = new StreamReader(_stream, leaveOpen: true);
var csvReader = new CsvReader(streamReader, CsvConfiguration);
using var csvReader = new CsvReader(new StreamReader(stream), CsvConfiguration, leaveOpen);

await csvReader.ReadAsync();
csvReader.ReadHeader();

_headerRaw = string.Join(csvReader.Configuration.Delimiter, csvReader.HeaderRecord);
HeaderRaw = string.Join(csvReader.Configuration.Delimiter, csvReader.HeaderRecord);

_totalCount = 0;
TotalCount = 0;

while (await csvReader.ReadAsync())
{
_totalCount++;
TotalCount++;
}

if (_stream.CanSeek)
if (stream.CanSeek)
{
_stream.Seek(streamPosition, SeekOrigin.Begin);
stream.Seek(streamPosition, SeekOrigin.Begin);
}

return _totalCount.Value;
return TotalCount.Value;
}

public virtual async Task<object[]> ReadNextPageAsync(ImportContext context)
Expand All @@ -110,7 +135,7 @@ public virtual async Task<object[]> ReadNextPageAsync(ImportContext context)
result.Add(new CsvImportRecord<TCsvImportable>
{
Row = row,
RawHeader = _headerRaw,
RawHeader = HeaderRaw,
RawRecord = rawRecord,
Record = record,
});
Expand Down Expand Up @@ -195,21 +220,30 @@ public void Dispose()

protected virtual void Dispose(bool disposing)
{
CsvReader.Dispose();
_stream?.Dispose();
if (_disposed)
{
return;
}
if (disposing)
{
CsvReader.Dispose();
Stream?.Dispose();
CountStream?.Dispose();
}
_disposed = true;
}

private CsvConfiguration MergeWithDefaultConfig(CsvConfiguration csvConfiguration, ImportContext context)
{
var defaultCsvConfiguration = GetConfiguration(context);
var result = csvConfiguration;
result.Delimiter = result.Delimiter ?? defaultCsvConfiguration.Delimiter;
result.PrepareHeaderForMatch = result.PrepareHeaderForMatch ?? defaultCsvConfiguration.PrepareHeaderForMatch;
result.BadDataFound = result.BadDataFound ?? defaultCsvConfiguration.BadDataFound;
result.ReadingExceptionOccurred = result.ReadingExceptionOccurred ?? defaultCsvConfiguration.ReadingExceptionOccurred;
result.MissingFieldFound = result.MissingFieldFound ?? defaultCsvConfiguration.MissingFieldFound;

return result;

csvConfiguration.Delimiter ??= defaultCsvConfiguration.Delimiter;
csvConfiguration.PrepareHeaderForMatch ??= defaultCsvConfiguration.PrepareHeaderForMatch;
csvConfiguration.BadDataFound ??= defaultCsvConfiguration.BadDataFound;
csvConfiguration.ReadingExceptionOccurred ??= defaultCsvConfiguration.ReadingExceptionOccurred;
csvConfiguration.MissingFieldFound ??= defaultCsvConfiguration.MissingFieldFound;

return csvConfiguration;
}
}
}
Loading