Extract text file from zip, parse content and post to azure queue storage

Question

Problem

I'm trying to create the most performant code to:

• Extract a text file from a zip
• parse its content
• serialize class
• post to Azure Queue Storage

Scenario

4 times per year I receive 20 zip files. All zip files have just one text file inside. Each text file is a fixed width file. Each line in the text file is a different type of data. There are 3 of them:

• Company (First character of line is "1")
• Partners (First character of line is "2")
• Activity (First character of line is "6")

They are always in sequence:

1First Company Data ..................................
2First Company First Partner .........................
2First Company Second Partner ........................
2First Company Third Partner .........................
6First Company Activity ..............................
1Second Company Data .................................
2Second Company First Partner ........................
2Second Company Second Partner .......................
2Second Company Third Partner ........................
2Second Company Fourth Partner .......................
6Second Company Activity ............................. 

There are always one line for company, one line for activity, but zero to many partners.

Activity is compound of 7 numbers sequence repeated multiple times. For example:

• 1111111 is an activity

if the company have more than one activity, they are putted in sequence like:

• 111111122222223333333

if there's no other activity, zeros are used until the end of line:

• 11111112222222333333300000000000000000000000000000000000

the line identifier for activity is "6", so, the full line stills like:

• 611111112222222333333300000000000000000000000000000000000

Two companys appear like this way:

1First Company Data .....................................
2First Company First Partner ............................
2First Company Second Partner ...........................
2First Company Third Partner ............................
611111112222222333333300000000000000000000000000000000000
1Second Company Data ....................................
2Second Company First Partner ...........................
2Second Company Second Partner ..........................
2Second Company Third Partner ...........................
2Second Company Fourth Partner ..........................
644444445555555000000000000000000000000000000000000000000

the zip files need to be read in sequence, because, the last line of a file may be in the middle of company data:

1First Company Data .....................................
2First Company First Partner ............................
2First Company Second Partner ...........................
2First Company Third Partner ............................
611111112222222333333300000000000000000000000000000000000
1Second Company Data ....................................
2Second Company First Partner ...........................

[------ END OF FILE 1 ------]

[----- BEGIN OF FILE 2 -----]

2Second Company Second Partner ..........................
2Second Company Third Partner ...........................
2Second Company Fourth Partner ..........................
644444445555555000000000000000000000000000000000000000000

About the code

I never worked with multithread before, I tried to put a thread to open the zip files, read its content, parse the data and post it to a Blocking Collection. The second thread I used to post Base 64 Encoded serialized class to Azure Queue Storage. I tried to use more than one thread to this but with no success. The third thread I use just to notify about the processing. I used a trick to post data to Azure Queue in batch, it sped up the performance, but I don't know if there is a more secure and better way to do this.

I'm posting to Azure Queue Storage, and, in the other side, I have another console to take data from Azure Queue and post to SQL Server. If there is a way to do this that is secure and reliable, but without using Azure Queue, I'll appreciate too. I tried this method in the first time, but network problems in the middle of process make me lost 120,000 companies data. The total of companies is about 43,000,000.

The data is public, and can be downloaded at: http://receita.economia.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/dados-publicos-cnpj

Code

this is the main console code

using Azure.Storage.Queues;
using BaseReceita.Producer.Model;
using ICSharpCode.SharpZipLib.Zip;
using Microsoft.Extensions.Configuration;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace BaseReceita.Producer
{
    class Program
    {
        private static string FilesDirectory { get; set; }

        static void Main(string[] args)
        {
            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Starting...");

            Console.WriteLine("Specify the folder with zip files: ");
            FilesDirectory = @"" + Console.ReadLine();

            Start().Wait();
        }

        private static async Task Start()
        {
            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] starting to read...");

            var watch = new Stopwatch();
            watch.Start();

            ServicePointManager.UseNagleAlgorithm = false;
            ServicePointManager.DefaultConnectionLimit = 1000;


            object lockobj = new object();
            long RegistrosProcessados = 0;
            var ts = new CancellationTokenSource();
            CancellationToken ct = ts.Token;

            IConfigurationRoot Configuration = new ConfigurationBuilder()
                                                .SetBasePath(Directory.GetCurrentDirectory())
                                                .AddJsonFile("appsettings.json")
                                                .Build();

            string storageConnectionString = Configuration["Storage:ConnectionString"];

            QueueClient queueClient = new QueueClient(storageConnectionString, "rfb-update-queue");

            //-------------------- Collection
            BufferBlock<string> buffer = new BufferBlock<string>(new DataflowBlockOptions() { BoundedCapacity = 50000 });

            //-------------------- Consumers
            var Consumers = new List<Task>();
            for (var i = 0; i < 1; i++)
            {
                Consumers.Add(Task.Run(async () => {
                    Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Consumer starting.");

                    while (await buffer.OutputAvailableAsync(ct))
                    {
                        if (buffer.TryReceiveAll(out var items))
                        {
                            try
                            {
                                await SendMessagesAsync(queueClient, items.AsEnumerable());

                                lock (lockobj)
                                    RegistrosProcessados = RegistrosProcessados + items.Count;

                            }
                            catch (Exception e)
                            {
                                Console.ForegroundColor = ConsoleColor.Red;
                                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Erro: {e.Message}");
                                Console.ResetColor();
                                //throw;
                            }
                        }
                    }

                    Console.ForegroundColor = ConsoleColor.DarkCyan;
                    Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Consumer finalized");
                    Console.ResetColor();
                }));
            }

            

            //-------------------- Notifier
            Task Notifier = Task.Factory.StartNew(() =>
            {


                while (true)
                {
                    if (!ct.IsCancellationRequested)
                    {
                        //F = Buffer Size
                        //P = Processed companies
                        //in the sequence, average processed per second, per minute and per hour

                        Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] F-{buffer.Count} P-{RegistrosProcessados:n0} ({RegistrosProcessados / watch.Elapsed.TotalSeconds:n0}/s | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60):n0}/m | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60 / 60):n0}/h)");
                        Thread.Sleep(5000); //notify every 5 seconds
                    }
                    else
                    {
                        break;
                    }
                }
            });


            //-------------------- Producer
            Task Producer = Task.Run(async () =>
            {
                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer started.");

                string conjunto = string.Empty;
                string linha = string.Empty;

                try
                {
                    //open all zip files
                    foreach (string file in Directory.EnumerateFiles(FilesDirectory, "*.zip"))
                    {
                        //open zip
                        using (ZipFile zf = new ZipFile(file))
                        {
                            //take all files (aways will be one file
                            foreach (ZipEntry entry in zf)
                            {
                                //open as stream
                                using (var stream = zf.GetInputStream(entry))
                                using (var reader = new StreamReader(stream))
                                {
                                    //read line from file
                                    while ((linha = reader.ReadLine()) != null)
                                    {
                                        string tipoCampo = linha.Substring(0, 1);

                                        if (tipoCampo == "1")
                                        {
                                            //every "1" is a new company, than, I parse the last company based on all the text extracted
                                            EmpresaModel empresa = Parse(conjunto);

                                            if (empresa != null)
                                            {
                                                //the first time will be null
                                                //the others wont
                                                //serialize, compress and post to buffer
                                                string json = JsonConvert.SerializeObject(empresa);
                                                string compressed = Base64Compress(json);
                                                buffer.Post(compressed);
                                            }

                                            conjunto = linha;
                                        }
                                        else if (tipoCampo != "0")
                                        {
                                            conjunto = conjunto + Environment.NewLine + linha;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
                catch (Exception e)
                {
                    Console.ForegroundColor = ConsoleColor.Red;
                    Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer Error: {e.Message}");
                    Console.ResetColor();
                }
                
                Console.ForegroundColor = ConsoleColor.DarkCyan;
                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Producer finalized");
                Console.ResetColor();
            });

            try
            {
                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Starting threads.");

                List<Task> tasks = new List<Task>();
                tasks.Add(Producer);
                tasks.AddRange(Consumers);
                
                Task.WaitAll(tasks.ToArray());

                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Threads finalized");
            }
            catch (Exception e)
            {
                Console.ForegroundColor = ConsoleColor.Red;
                Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Error: {e.Message}");
                Console.ResetColor();
            }

            ts.Cancel();

            watch.Stop();

            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Total Time: {watch.Elapsed.ToString()}");
            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] Pushed-{RegistrosProcessados:n0} ({RegistrosProcessados / watch.Elapsed.TotalSeconds:n0}/s | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60):n0}/m | {RegistrosProcessados / (watch.Elapsed.TotalSeconds / 60 / 60):n0}/h)");

            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}] End of process");
            Console.ReadLine();
        }

        private static EmpresaModel Parse(string conjunto)
        {
            EmpresaModel empresa = null;

            if (!string.IsNullOrEmpty(conjunto))
            {
                string[] linhas = conjunto.Trim().Split(new[] { '\n' });

                foreach (string linha in linhas)
                {
                    string cnpj = linha.Substring(3, 14);

                    if (linha.Substring(0, 1) == "1")
                    {
                        //Company

                        empresa = new EmpresaModel();

                        empresa.Cnpj = cnpj;
                        empresa.IndicadorMatrizFilial = linha.Substring(17, 1).Trim();
                        empresa.RazaoSocial = linha.Substring(18, 150).Trim();
                        empresa.NomeFantasia = linha.Substring(168, 55).Trim();
                        empresa.CodigoSituacaoCadastral = linha.Substring(223, 2).Trim();
                        //empresa.SituacaoCadastral = (string)SituacaoCadastral.FirstOrDefault(x => x.Key == empresa.CodigoSituacaoCadastral).Value;
                        empresa.DataSituacaoCadastral = linha.Substring(225, 8).Trim();
                        empresa.CodigoMotivoSituacaoCadastral = linha.Substring(233, 2).Trim();
                        //empresa.MotivoSituacaoCadastral = (string)MotivoSituacaoCadastral.FirstOrDefault(x => x.Key == empresa.CodigoMotivoSituacaoCadastral).Value;
                        empresa.CidadeExterior = linha.Substring(235, 55).Trim();
                        empresa.CodigoPais = linha.Substring(290, 3).Trim();
                        empresa.Pais = linha.Substring(293, 70).Trim();
                        empresa.CodigoNaturezaJuridica = linha.Substring(363, 3).Trim() + "-" + linha.Substring(366, 1).Trim();
                        //empresa.NaturezaJuridica = (string)NaturezaJuridica.FirstOrDefault(x => x.Key == empresa.CodigoNaturezaJuridica).Value;
                        empresa.DataInicioAtividade = linha.Substring(367, 8).Trim();
                        empresa.IdCnae = linha.Substring(375, 7).Trim();
                        empresa.TipoLogradouro = linha.Substring(382, 20).Trim();
                        empresa.Logradouro = linha.Substring(402, 60).Trim();
                        empresa.Numero = linha.Substring(462, 6).Trim();
                        empresa.Complemento = linha.Substring(468, 156).Trim();
                        empresa.Bairro = linha.Substring(624, 50).Trim();
                        empresa.Cep = linha.Substring(674, 8).Trim();
                        empresa.UF = linha.Substring(682, 2).Trim();
                        empresa.CodigoMunicipio = linha.Substring(684, 4).Trim();
                        empresa.Municipio = linha.Substring(688, 50).Trim();
                        empresa.DDD1 = linha.Substring(738, 4).Trim();
                        empresa.Telefone1 = linha.Substring(742, 8).Trim();
                        empresa.DDD2 = linha.Substring(750, 4).Trim();
                        empresa.Telefone2 = linha.Substring(754, 8).Trim();
                        empresa.DDDFax = linha.Substring(762, 4).Trim();
                        empresa.TelefoneFax = linha.Substring(766, 8).Trim();
                        empresa.Email = linha.Substring(774, 115).Trim();
                        empresa.CodigoQualificacaoResponsavel = linha.Substring(889, 2).Trim();
                        empresa.CapitalSocial = linha.Substring(891, 14).Trim();
                        empresa.CodigoPorteEmpresa = linha.Substring(905, 2).Trim();
                        empresa.CodigoOpcaoSimplesNacional = linha.Substring(907, 1).Trim();
                        empresa.DataOpcaoSimples = linha.Substring(908, 8).Trim();
                        empresa.OptanteMei = linha.Substring(924, 1).Trim();
                        empresa.SituacaoEspecial = linha.Substring(925, 23).Trim();
                        empresa.DataSituacaoEspecial = linha.Substring(948, 8).Trim();
                    }
                    else if (linha.Substring(0, 1) == "2")
                    {
                        //Partners

                        QuadroSocietarioModel qsa = new QuadroSocietarioModel();

                        qsa.Cnpj = linha.Substring(3, 14).Trim();
                        qsa.IdentificadorSocio = linha.Substring(17, 1).Trim();
                        qsa.NomeSocio = linha.Substring(18, 150).Trim();
                        qsa.CnpjCpfSocio = linha.Substring(168, 14).Trim();
                        qsa.CodigoQualificacaoSocio = linha.Substring(182, 2).Trim();
                        //qsa.QualificacaoSocio = (string)QualificacaoResponsavelSocio.FirstOrDefault(x => x.Key == qsa.CodigoQualificacaoSocio).Value;
                        qsa.PercentualCapitalSocial = linha.Substring(184, 5).Trim();
                        qsa.DataEntradaSociedade = linha.Substring(189, 8).Trim();
                        qsa.CodigoPais = linha.Substring(197, 3).Trim();
                        qsa.Pais = linha.Substring(200, 70).Trim();
                        qsa.CpfRepresentanteLegal = linha.Substring(270, 11).Trim();
                        qsa.NomeRepresentante = linha.Substring(281, 60).Trim();
                        qsa.CodigoQualificacaoRepresentanteLegal = linha.Substring(341, 2).Trim();

                        empresa?.QuadroSocietario.Add(qsa);
                    }
                    else if (linha.Substring(0, 1) == "6")
                    {
                        //Activity
                        string[] cnaes =
                            Split(linha.Substring(17, 693).Trim(), 7)
                            .Where(x => x != "0000000")
                            .Where(x => !string.IsNullOrEmpty(x.Trim()))
                            //.Select(x => "cnae/" + x)
                            .ToArray();

                        foreach (string cnae in cnaes)
                        {
                            CnaeSecundarioModel cnaeSecundario = new CnaeSecundarioModel();

                            cnaeSecundario.Cnpj = cnpj;
                            cnaeSecundario.Cnae = cnae;

                            empresa?.CnaesSecundarios.Add(cnaeSecundario);
                        }
                    }
                }
            }

            return empresa;
        }

        private static IEnumerable<string> Split(string str, int chunkSize)
        {
            return Enumerable.Range(0, str.Length / chunkSize)
                .Select(i => str.Substring(i * chunkSize, chunkSize));
        }

        private static string Base64Compress(string s)
        {
            byte[] inputBytes = Encoding.UTF8.GetBytes(s);

            using (var outputStream = new MemoryStream())
            {
                using (var gZipStream = new System.IO.Compression.GZipStream(outputStream, System.IO.Compression.CompressionMode.Compress))
                    gZipStream.Write(inputBytes, 0, inputBytes.Length);

                var outputBytes = outputStream.ToArray();

                var outputbase64 = Convert.ToBase64String(outputBytes);
                return outputbase64;
            }
        }

        public static async Task SendMessagesAsync(QueueClient queue, IEnumerable<string> messages)
        {
            await Task.WhenAll(
                    from partition in Partitioner.Create(messages).GetPartitions(500)
                    select Task.Run(async delegate
                    {
                        using (partition)
                            while (partition.MoveNext())
                                await queue.SendMessageAsync(partition.Current);
                    }));
        }
    }
}

here are the entities

using System.Collections.Generic;

namespace BaseReceita.Producer.Model
{
    public class EmpresaModel
    {
        public EmpresaModel()
        {
            QuadroSocietario = new HashSet<QuadroSocietarioModel>();
            CnaesSecundarios = new HashSet<CnaeSecundarioModel>();
        }

        public string Cnpj { get; set; }
        public string IndicadorMatrizFilial { get; set; }

        public string RazaoSocial { get; set; }

        public string NomeFantasia { get; set; }
        public string CodigoSituacaoCadastral { get; set; }
        public string DataSituacaoCadastral { get; set; }
        public string CodigoMotivoSituacaoCadastral { get;  set; }
        public string CidadeExterior { get;  set; }
        public string CodigoPais { get;  set; }
        public string Pais { get;  set; }
        public string CodigoNaturezaJuridica { get;  set; }
        public string DataInicioAtividade { get; set; }
        public string IdCnae { get; set; }
        public string TipoLogradouro { get;  set; }
        public string Logradouro { get;  set; }
        public string Numero { get;  set; }
        public string Complemento { get;  set; }
        public string Bairro { get;  set; }
        public string Cep { get;  set; }
        public string UF { get;  set; }
        public string CodigoMunicipio { get;  set; }
        public string Municipio { get;  set; }
        public string DDD1 { get;  set; }
        public string Telefone1 { get;  set; }
        public string DDD2 { get;  set; }
        public string Telefone2 { get;  set; }
        public string DDDFax { get;  set; }
        public string TelefoneFax { get;  set; }
        public string Email { get;  set; }
        public string CodigoQualificacaoResponsavel { get;  set; }
        public string CapitalSocial { get; set; }
        public string CodigoPorteEmpresa { get;  set; }
        public string CodigoOpcaoSimplesNacional { get;  set; }
        public string DataOpcaoSimples { get; set; }
        public string OptanteMei { get;  set; }
        public string SituacaoEspecial { get;  set; }
        public string DataSituacaoEspecial { get;  set; }

        public HashSet<QuadroSocietarioModel> QuadroSocietario { get; set; }
        public HashSet<CnaeSecundarioModel> CnaesSecundarios { get; set; }
    }

    public class QuadroSocietarioModel
    {
        public string Cnpj { get; set; }
        public string IdentificadorSocio { get; set; }
        public string NomeSocio { get; set; }
        public string CnpjCpfSocio { get; set; }
        public string CodigoQualificacaoSocio { get;  set; }
        public string PercentualCapitalSocial { get;  set; }
        public string DataEntradaSociedade { get;  set; }
        public string CodigoPais { get;  set; }
        public string Pais { get;  set; }
        public string CpfRepresentanteLegal { get;  set; }
        public string NomeRepresentante { get;  set; }
        public string CodigoQualificacaoRepresentanteLegal { get;  set; }
    }

    public class CnaeSecundarioModel
    {
        public string Cnpj { get; set; }
        public string Cnae { get; set; }
    }
}

Answer 1

I assume you are using .net core. If so you should make the Main method to be async Task Main(string[] args) have been supported since .net core 2.0. I would move the configuration builder into the main method as well. Have everything you need to support running your app in the main method.

Right now you have a couple of big methods that do a lot of things and we want to have more methods but each method do one thing. Some simple ones to break out

/// <summary>
/// Creates Empresa (Company)
/// </summary>
/// <param name="data">Info to use to fill in model (fixed width)</param>
/// <returns></returns>
private EmpresaModel CreateCompany(string data)
{
    return new EmpresaModel()
    {
        Cnpj = data.Substring(3, 14),
        IndicadorMatrizFilial = data.Substring(17, 1).Trim(),
        RazaoSocial = data.Substring(18, 150).Trim(),
        NomeFantasia = data.Substring(168, 55).Trim(),
        CodigoSituacaoCadastral = data.Substring(223, 2).Trim(),
        DataSituacaoCadastral = data.Substring(225, 8).Trim(),
        CodigoMotivoSituacaoCadastral = data.Substring(233, 2).Trim(),
        CidadeExterior = data.Substring(235, 55).Trim(),
        CodigoPais = data.Substring(290, 3).Trim(),
        Pais = data.Substring(293, 70).Trim(),
        CodigoNaturezaJuridica = data.Substring(363, 3).Trim() + "-" + data.Substring(366, 1).Trim(),
        DataInicioAtividade = data.Substring(367, 8).Trim(),
        IdCnae = data.Substring(375, 7).Trim(),
        TipoLogradouro = data.Substring(382, 20).Trim(),
        Logradouro = data.Substring(402, 60).Trim(),
        Numero = data.Substring(462, 6).Trim(),
        Complemento = data.Substring(468, 156).Trim(),
        Bairro = data.Substring(624, 50).Trim(),
        Cep = data.Substring(674, 8).Trim(),
        UF = data.Substring(682, 2).Trim(),
        CodigoMunicipio = data.Substring(684, 4).Trim(),
        Municipio = data.Substring(688, 50).Trim(),
        DDD1 = data.Substring(738, 4).Trim(),
        Telefone1 = data.Substring(742, 8).Trim(),
        DDD2 = data.Substring(750, 4).Trim(),
        Telefone2 = data.Substring(754, 8).Trim(),
        DDDFax = data.Substring(762, 4).Trim(),
        TelefoneFax = data.Substring(766, 8).Trim(),
        Email = data.Substring(774, 115).Trim(),
        CodigoQualificacaoResponsavel = data.Substring(889, 2).Trim(),
        CapitalSocial = data.Substring(891, 14).Trim(),
        CodigoPorteEmpresa = data.Substring(905, 2).Trim(),
        CodigoOpcaoSimplesNacional = data.Substring(907, 1).Trim(),
        DataOpcaoSimples = data.Substring(908, 8).Trim(),
        OptanteMei = data.Substring(924, 1).Trim(),
        SituacaoEspecial = data.Substring(925, 23).Trim(),
        DataSituacaoEspecial = data.Substring(948, 8).Trim(),
    };
}

/// <summary>
/// Creates QuadroSocietario (Partner)
/// </summary>
/// <param name="data">Info to use to fill in model (fixed width)</param>
/// <returns></returns>
private QuadroSocietarioModel CreatePartner(string data)
{
    return new QuadroSocietarioModel()
    {
        Cnpj = data.Substring(3, 14).Trim(),
        IdentificadorSocio = data.Substring(17, 1).Trim(),
        NomeSocio = data.Substring(18, 150).Trim(),
        CnpjCpfSocio = data.Substring(168, 14).Trim(),
        CodigoQualificacaoSocio = data.Substring(182, 2).Trim(),
        PercentualCapitalSocial = data.Substring(184, 5).Trim(),
        DataEntradaSociedade = data.Substring(189, 8).Trim(),
        CodigoPais = data.Substring(197, 3).Trim(),
        Pais = data.Substring(200, 70).Trim(),
        CpfRepresentanteLegal = data.Substring(270, 11).Trim(),
        NomeRepresentante = data.Substring(281, 60).Trim(),
        CodigoQualificacaoRepresentanteLegal = data.Substring(341, 2).Trim(),
    };
}

/// <summary>
/// Creates CnaeSecundarioModel (Activities)
/// </summary>
/// <param name="data">Info to use to fill in model (fixed width)</param>
/// <returns></returns>
private IEnumerable<CnaeSecundarioModel> CreateActivities(string data)
{
    var cnpj = data.Substring(3, 14);
    // why do we start at 17?
    return Split(data.Substring(17, 693).Trim(), 7)
         .Where(x => !string.IsNullOrEmpty(x) && x != "0000000")
         .Select(cnae => new CnaeSecundarioModel()
         {
             Cnae = cnae,
             Cnpj = cnpj
         });
}

To help "hide" the magic values for Company/Partners/Activities we can create an enum for those values. Also a value for unknown and end of file which we will use in a bit

public enum LineType
{
    Skip = '0',
    Company = '1',
    Partners = '2',
    Activity = '6',
    EOF = 'E',
    Unknown = 'X'
}

Since we are using TPL DataFlow we can create a mesh that will help process. So first thing we need is a method to convert the zip file into models and a method to read the entries in the zip file. I'm using System.IO.Compression for reading the zip and Microsoft.Extensions.Logging to add some logging.

/// <summary>
/// Converts Fixed Line files into Company models
/// </summary>
/// <param name="lines">Lines from file</param>
/// <param name="token">Cancellation Token</param>
/// <returns></returns>
private async IAsyncEnumerable<EmpresaModel> Deserialize(string file, [EnumeratorCancellation] CancellationToken token = default)
{
    EmpresaModel empresa = null;
    await foreach (var line in GetData(file).WithCancellation(token).ConfigureAwait(false))
    {
        if (string.IsNullOrWhiteSpace(line))
        {
            continue;
        }
        var type = (LineType)line[0];
        switch (type)
        {
            case LineType.EOF:
                {
                    if (empresa != null)
                    {
                        yield return empresa;
                        empresa = null;
                    }
                    break;
                }
            case LineType.Skip:
                {
                    break;
                }
            case LineType.Company:
                {
                    if (empresa != null)
                    {
                        yield return empresa;
                    }

                    empresa = CreateCompany(line);
                    break;
                }
            case LineType.Partners:
                {
                    if (empresa == null)
                    {
                        this.logger.LogWarning(new EventId((int)LineType.Partners, LineType.Partners.ToString()), "Missing Company");
                        break;
                    }
                    empresa.QuadroSocietario.Add(CreatePartner(line));
                    break;
                }
            case LineType.Activity:
                {
                    if (empresa == null)
                    {
                        this.logger.LogWarning(new EventId((int)LineType.Activity, LineType.Activity.ToString()), "Missing Company");
                        break;
                    }
                    foreach (var activity in CreateActivities(line))
                    {
                        empresa.CnaesSecundarios.Add(activity);
                    }
                    break;
                }
            default:
                {
                    this.logger.LogError(new EventId((int)LineType.Unknown, LineType.Unknown.ToString()), new FileFormatException("Unkown line type"), "Unkown line type");
                    break;
                }
        }
    }

    if (empresa != null)
    {
        yield return empresa;
    }
}

/// <summary>
/// Open zip files reads all files and outputs their text
/// </summary>
/// <param name="zipFile"></param>
/// <param name="token"></param>
/// <returns>Enumerable for each file in archive with asyncenum to read the lines in that file</returns>
private async IAsyncEnumerable<string> GetData(string zipFile, [EnumeratorCancellation] CancellationToken token = default)
{
    using (var archive = ZipFile.OpenRead(zipFile))
    {
        foreach (var file in archive.Entries)
        {
            using (var fileStream = file.Open())
            {
                using (var reader = new StreamReader(fileStream))
                {
                    while (!reader.EndOfStream && !token.IsCancellationRequested)
                    {
                        var line = await reader.ReadLineAsync().ConfigureAwait(false);
                        if (line != null)
                        {
                            yield return line;
                        }
                    }
                    // special case for end of file
                    yield return ((Char)LineType.EOF).ToString();
                }
            }
        }
    }
}

Now we need a custom Data flow block that will take in path to zipfile and output all the models in it.

/// <summary>
/// Creates a Data Block that takes in the zip file path and out put models
/// </summary>
/// <param name="ExecutionDataflowBlockOptions"></param>
/// <returns>Custom Data Flow Block</returns>
private IPropagatorBlock<string, EmpresaModel> ExtractZip(ExecutionDataflowBlockOptions options = null)
{
    var token = options?.CancellationToken ?? CancellationToken.None;
    // this will Broadcase out the models once build
    var source = new TransformBlock<EmpresaModel, EmpresaModel>(t => t, options);
    // Will go threw the zip and create the models
    var target = new ActionBlock<string>(async file =>
    {
        await foreach (var model in Deserialize(file).WithCancellation(token).ConfigureAwait(false))
        {
            await source.SendAsync(model, token).ConfigureAwait(false);
        }
    }, options);

    // When the target is set to the completed state set the source to the completed state.
    target.Completion.ContinueWith(_ => source.Complete());

    return DataflowBlock.Encapsulate(target, source);
}

For outputting progress I typically use the IProgress<> interface. Because I want it to be threadsafe I'm going to implement the interface myself and not use the Progress class.

public class Notifier : IProgress<int> 
{
    private int totalCount = 0;
    private DateTime startTime = DateTime.Now;
    private DateTime lastNotified = DateTime.Now.Subtract(TimeSpan.FromSeconds(5));
    public void Report(int numberToAdd)
    {
        var total = Interlocked.Add(ref totalCount, numberToAdd);
        if (DateTime.Now.Subtract(lastNotified) >= TimeSpan.FromSeconds(5))
        {
            var totalSeconds = DateTime.Now.Subtract(startTime).TotalSeconds;
            Console.WriteLine($"[{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")}]  P-{total:n0} ({total / totalSeconds:n0}/s | {total / (totalSeconds / 60):n0}/m | {total / (totalSeconds / 60 / 60):n0}/h)");
            lastNotified = DateTime.Now;
        }
    }
}

We will create a method to encode the models. I'm using the System.Text.Json and pushing json stream into the gzip stream to not have to create a memory stream

private async Task<string> SerializeAsync(EmpresaModel model, CancellationToken token)
{
    using (var memoryStream = new MemoryStream())
    {
        using (var gzipStream = new GZipStream(memoryStream, CompressionMode.Compress))
        {
            await JsonSerializer.SerializeAsync(gzipStream, model, null, token).ConfigureAwait(false);
        }
        return Convert.ToBase64String(memoryStream.ToArray());
    }
}

The last thing we need is a method to send to the Azure. If wanting to go to SQL and not have issue where you lost records then should look into Poly to handle transient errors. Plus wrap it all in a transaction so they either complete or rollback as one statement. With this when Poly retries you will get atomic writes

private async Task<string> SendToQueue(QueueClient client, string message, CancellationToken token)
{
    // if want to go directly to SQL then in this method can add Poly to handle transient errors
    var receipt = await client.SendMessageAsync(message, token).ConfigureAwait(false);
    return receipt.Value.MessageId;
}

Noe that we have all the methods we just need to create the mesh pipeline.

public async Task Start(string directory, QueueClient client, IProgress<int> progress, CancellationToken token)
{
    var executionBlockOptions = new ExecutionDataflowBlockOptions()
    {
        CancellationToken = token,
        //   MaxDegreeOfParallelism = 2,
        BoundedCapacity = 500
    };

    var extractZip = ExtractZip(executionBlockOptions);
    var encode = new TransformBlock<EmpresaModel, string>(async x => await SerializeAsync(x, token).ConfigureAwait(false), executionBlockOptions);
    var sendToQueue = new TransformBlock<string, string>(async x => await SendToQueue(client, x, token).ConfigureAwait(false), executionBlockOptions);
    var report = new ActionBlock<string>(_ => progress.Report(1), executionBlockOptions);
    var linkOptions = new DataflowLinkOptions()
    {
        PropagateCompletion = true,
    };
    extractZip.LinkTo(encode, linkOptions);
    encode.LinkTo(sendToQueue, linkOptions);
    sendToQueue.LinkTo(report, linkOptions);

    foreach (var file in Directory.EnumerateFiles(directory, "*.zip"))
    {
        await extractZip.SendAsync(file).ConfigureAwait(false);
    }
    extractZip.Complete();
    await report.Completion.ConfigureAwait(false);
}

With all the async work we doing it actually slowed down how fast my machine could do if I set MaxDegreeOfParallelism. You could also have each Data flow block have its own execution option and tinker to see what performs best on your machine/network. Basically we setup the mesh to extract the data, then encode the data then sent to azure and finally report the progress. Then once the mesh is setup we loop through all the zip files in the directory and push the value into the mesh then wait for the entire mesh to finish.

Every machine is different but I downloaded 6 of the zips and this used ~95% of my 8 core and processed around 7,500 companies a second. You can always tweak the data flow options to see what works best as I just took some guess, to be honest this took a lot of time but I was intrigued about it. Using the IAsyncEnumerable will help lower the memory as we don't need to load as much of the file into memory.