Dart: Otimização e Implantação de Projeto — DataPipeline

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

import 'dart:isolate';

// Ruim: Fragmentação estática - trabalho desigual
List<List<T>> staticChunk<T>(List<T> data, int chunks) {
  final size = (data.length / chunks).ceil();
  return [
    for (var i = 0; i < chunks; i++)
      data.sublist(i * size, ((i + 1) * size).clamp(0, data.length))
  ];
}

// Bom: Work stealing dinâmico com SendPort
class DynamicIsolatePool {
  final int _poolSize;
  var _pendingChunks = <List<Map<String, dynamic>>>[];
  final _results = <AnalysisResult>[];
  var _activeWorkers = 0;
  late Completer<List<AnalysisResult>> _completer;

  DynamicIsolatePool({int poolSize = 4}) : _poolSize = poolSize;

  Future<List<AnalysisResult>> process(
    List<Map<String, dynamic>> allOrders, {
    int chunkSize = 50000,
  }) async {
    _pendingChunks = [
      for (var i = 0; i < allOrders.length; i += chunkSize)
        allOrders.sublist(i, (i + chunkSize).clamp(0, allOrders.length))
    ];

    _completer = Completer<List<AnalysisResult>>();
    _activeWorkers = 0;

    final workers = <_Worker>[];
    for (var i = 0; i < _poolSize && _pendingChunks.isNotEmpty; i++) {
      final chunk = _pendingChunks.removeAt(0);
      workers.add(await _spawnWorker(chunk));
      _activeWorkers++;
    }

    return _completer.future;
  }

  Future<_Worker> _spawnWorker(List<Map<String, dynamic>> initialChunk) async {
    final receivePort = ReceivePort();
    final isolate = await Isolate.spawn(_workerEntry, receivePort.sendPort);
    final sendPort = await receivePort.first as SendPort;

    final responsePort = ReceivePort();
    sendPort.send(_WorkMessage(data: initialChunk, responsePort: responsePort.sendPort));

    responsePort.listen((result) {
      _results.add(result as AnalysisResult);
      _activeWorkers--;

      if (_pendingChunks.isNotEmpty) {
        final nextChunk = _pendingChunks.removeAt(0);
        sendPort.send(_WorkMessage(data: nextChunk, responsePort: responsePort.sendPort));
        _activeWorkers++;
      } else if (_activeWorkers == 0) {
        isolate.kill(priority: Isolate.immediate);
        receivePort.close();
        responsePort.close();
        _completer.complete(_results);
      }
    });

    return _Worker(isolate: isolate, sendPort: sendPort);
  }

  static void _workerEntry(SendPort mainSendPort) {
    final receivePort = ReceivePort();
    mainSendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message is _WorkMessage) {
        final orders = message.data.map((m) => Order(
          id: m['id'] as String,
          amount: (m['amount'] as num).toDouble(),
          status: m['status'] as String,
          category: m['category'] as String,
        )).toList();

        final result = const OrderAnalyzer().analyze(orders);
        message.responsePort.send(result);
      }
    });
  }
}

class _Worker {
  final Isolate isolate;
  final SendPort sendPort;
  const _Worker({required this.isolate, required this.sendPort});
}

class _WorkMessage {
  final List<Map<String, dynamic>> data;
  final SendPort responsePort;
  const _WorkMessage({required this.data, required this.responsePort});
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

import 'dart:async';

class BackpressurePipeline {
  final DataSource _source;
  final OrderParser _parser;
  final int _maxConcurrent;

  BackpressurePipeline({
    required DataSource source,
    required OrderParser parser,
    int maxConcurrent = 1000,
  })  : _source = source,
        _parser = parser,
        _maxConcurrent = maxConcurrent;

  Future<AnalysisResult> execute() async {
    final orders = <Order>[];
    var pending = 0;
    final completer = Completer<void>();

    final subscription = _source.readLines().listen(
      (line) {
        try {
          orders.add(_parser.parse(line));
        } on FormatException {
          return;
        }

        pending++;
        if (pending >= _maxConcurrent) {
          subscription.pause();
        }
      },
      onDone: () {
        completer.complete();
      },
      onError: (e) {
        completer.completeError(e);
      },
    );

    await completer.future;

    return const OrderAnalyzer().analyze(orders);
  }

  // Stream pausável com controle de buffer
  Stream<Order> streamWithBackpressure() async* {
    var buffer = <Order>[];

    await for (final line in _source.readLines()) {
      try {
        buffer.add(_parser.parse(line));
      } on FormatException {
        continue;
      }

      if (buffer.length >= _maxConcurrent) {
        for (final order in buffer) {
          yield order;
        }
        buffer = [];
      }
    }

    for (final order in buffer) {
      yield order;
    }
  }
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

import 'dart:math';

class MemoryOptimizer {
  // Usar iterables em vez de lists quando possível
  static Iterable<Order> lazyParse(Iterable<String> lines, OrderParser parser) sync* {
    for (final line in lines) {
      try {
        yield parser.parse(line);
      } on FormatException {
        continue;
      }
    }
  }

  // Processamento em chunks para limitar memória
  static Future<AnalysisResult> processInChunks(
    Stream<String> lines,
    OrderParser parser, {
    int chunkSize = 100000,
  }) async {
    var chunk = <Order>[];
    var totalResult = _EmptyAnalysisResult();

    await for (final line in lines) {
      try {
        chunk.add(parser.parse(line));
      } on FormatException {
        continue;
      }

      if (chunk.length >= chunkSize) {
        final partial = const OrderAnalyzer().analyze(chunk);
        totalResult = totalResult.merge(partial);
        chunk = [];
      }
    }

    if (chunk.isNotEmpty) {
      final partial = const OrderAnalyzer().analyze(chunk);
      totalResult = totalResult.merge(partial);
    }

    return totalResult.toAnalysisResult();
  }
}

class _EmptyAnalysisResult {
  int totalOrders = 0;
  int completedOrders = 0;
  double revenue = 0;
  double tax = 0;
  final Map<String, double> revenueByCategory = {};

  _EmptyAnalysisResult merge(AnalysisResult other) {
    totalOrders += other.totalOrders;
    completedOrders += other.completedOrders;
    revenue += other.revenue;
    tax += other.tax;
    for (final entry in other.revenueByCategory.entries) {
      revenueByCategory.update(entry.key, (v) => v + entry.value, ifAbsent: () => entry.value);
    }
    return this;
  }

  AnalysisResult toAnalysisResult() => AnalysisResult(
    totalOrders: totalOrders,
    completedOrders: completedOrders,
    revenue: revenue,
    tax: tax,
    total: revenue + tax,
    revenueByCategory: Map.unmodifiable(revenueByCategory),
  );
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

# Build debug (grande, inclui informações de debug)
dart compile exe bin/data_pipeline.dart -o build/data_pipeline_debug

# Build release (otimizado, menor)
dart compile exe bin/data_pipeline.dart -o build/data_pipeline \
  --define=MODE=release

# Verificar tamanho do binário
ls -lh build/

# Remover símbolos de debug (redução adicional de tamanho)
# No Linux/macOS:
strip build/data_pipeline

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

// Imports condicionais para tree shaking
const _isRelease = bool.fromEnvironment('MODE.release');

void log(String message) {
  if (!_isRelease) {
    print('[DEBUG] $message');
  }
}

// Esta classe inteira será removida por tree shaking no modo release
class DebugLogger {
  void log(String message) {
    if (!_isRelease) {
      print('[DEBUG] ${DateTime.now()}: $message');
    }
  }
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

# .github/workflows/ci.yml
name: DataPipeline CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  analyze-and-test:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        sdk: [stable, beta]

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1
        with:
          sdk: ${{ matrix.sdk }}

      - name: Instalar dependências
        run: dart pub get

      - name: Verificar formatação
        run: dart format --output=none --set-exit-if-changed .

      - name: Analisar código
        run: dart analyze --fatal-infos

      - name: Executar testes
        run: dart test --coverage=coverage

      - name: Enviar cobertura
        uses: codecov/codecov-action@v3
        with:
          files: coverage/lcov.info

  build-and-release:
    needs: analyze-and-test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Instalar dependências
        run: dart pub get

      - name: Compilar binário AOT
        run: |
          dart compile exe bin/data_pipeline.dart -o build/data_pipeline

      - name: Executar teste de integração
        run: |
          ./build/data_pipeline analyze --source csv --input test/fixtures/sample.csv

      - name: Enviar artefato binário
        uses: actions/upload-artifact@v4
        with:
          name: data_pipeline-linux
          path: build/data_pipeline

  publish:
    needs: build-and-release
    runs-on: ubuntu-latest
    if: startsWith(github.ref, 'refs/tags/v')

    steps:
      - uses: actions/checkout@v4

      - uses: dart-lang/setup-dart@v1

      - name: Instalar dependências
        run: dart pub get

      - name: Publicar no pub.dev
        run: dart pub publish --force
        env:
          PUB_CREDENTIALS: ${{ secrets.PUB_CREDENTIALS }}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

name: data_pipeline
description: Ferramenta CLI de análise de dados de e-commerce - processa pedidos em nível de milhão com Stream + Isolate
version: 1.0.0

homepage: https://github.com/bob/datapipeline
repository: https://github.com/bob/datapipeline
issue_tracker: https://github.com/bob/datapipeline/issues

environment:
  sdk: ^3.0.0

dependencies:
  args: ^2.4.2
  http: ^1.1.0
  csv: ^6.0.0

dev_dependencies:
  test: ^1.24.0
  build_runner: ^2.4.0
  json_serializable: ^6.7.0

executables:
  data_pipeline: data_pipeline

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

# CHANGELOG


---

## 7. Notas de Release

### (1) 1.0.0 - 2025-01-15

#### Funcionalidades
- Suporte a fonte de dados CSV e JSON
- Pipeline baseado em Stream com backpressure
- Pool de Isolates para processamento paralelo
- CLI com subcomandos analyze/export/validate
- Arquitetura Pattern Matching + classe Sealed

### (2) Desempenho
- 1M pedidos processados em ~8 segundos (4 núcleos)
- Uso de memória abaixo de 200MB para 1M registros

### (3) Testes
- 95% de cobertura de código
- Testes unitários + integração + E2E

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

# Checklist de pré-publicação

# 1. Publicação dry run
dart pub publish --dry-run

# 2. Verificar pontuação no pub.dev
dart pub publish --dry-run 2>&1 | grep -E "score|warning|error"

# 3. Gerar docs da API
dart doc .

# 4. Verificar conteúdo do pacote
dart pub pack  # cria .tar.gz para inspeção

# 5. Publicação final
dart pub publish

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

/// Analisa dados de pedidos de e-commerce e produz resultados estatísticos.
///
/// Use [OrderAnalyzer] como ponto de entrada principal para análise de dados.
/// Configure a [taxRate] com base na região de destino.
///
/// Exemplo:
/// ```dart
/// final analyzer = OrderAnalyzer(taxRate: 0.08);
/// final result = analyzer.analyze(orders);
/// print('Receita: \$${result.revenue} USD');
/// ```
///
/// Veja também:
/// - [AnalysisResult] para a estrutura de saída
/// - [Order] para o modelo de entrada
class OrderAnalyzer {
  /// Cria um analisador com a [taxRate] fornecida.
  ///
  /// Padrão é 0.08 (taxa de imposto dos EUA de 8%).
  /// Use taxas específicas por região: US=0.08, EU=0.20, UK=0.15.
  const OrderAnalyzer({this.taxRate = 0.08});

  /// A taxa de imposto aplicada a pedidos completados.
  final double taxRate;

  /// Analisa uma lista de [orders] e retorna estatísticas agregadas.
  ///
  /// Apenas pedidos com status 'completed' são incluídos nos cálculos
  /// de receita. Pedidos com outros status são contados em
  /// [AnalysisResult.totalOrders] mas excluídos da receita.
  AnalysisResult analyze(List<Order> orders) {
    // ... implementação
  }
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.

// ============================================
// Benchmark de Desempenho do DataPipeline
// Teste de Stress End-to-End da Alice
// ============================================

import 'dart:async';
import 'dart:io';

class Benchmark {
  final String name;
  final Stopwatch _stopwatch;

  Benchmark(this.name) : _stopwatch = Stopwatch();

  Future<T> run<T>(Future<T> Function() action) async {
    _stopwatch.reset();
    _stopwatch.start();
    final result = await action();
    _stopwatch.stop();
    print('$name: ${_stopwatch.elapsedMilliseconds}ms');
    return result;
  }
}

// Gerar dados de teste
List<String> generateOrders(int count) {
  final categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final regions = ['US', 'EU', 'UK', 'JP'];

  return List.generate(count, (i) {
    final category = categories[i % categories.length];
    final status = statuses[i % statuses.length];
    final region = regions[i % regions.length];
    final amount = (100 + (i * 37) % 5000).toDouble();
    return 'ORD-${(i + 1).toString().padLeft(7, '0')},$amount,$status,$category,$region';
  });
}

void main() async {
  print('=== Benchmark de Desempenho do DataPipeline ===\n');

  final sizes = [10000, 100000, 1000000];

  for (final size in sizes) {
    final label = size >= 1000000 ? '${size ~/ 1000000}M' : '${size ~/ 1000}K';
    print('--- $label pedidos ---');

    // Gerar e salvar em arquivo temporário
    final data = generateOrders(size);
    final tempFile = File('${Directory.systemTemp.path}/bench_${label}.csv');
    await tempFile.writeAsString(data.join('\n'));

    // Benchmark 1: Processamento Stream
    final streamResult = await Benchmark('Stream ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      final parser = const CsvOrderParser();
      final pipeline = Pipeline(source: source, parser: parser);
      return pipeline.execute();
    });

    // Benchmark 2: Processamento em chunks com eficiência de memória
    final chunkedResult = await Benchmark('Chunked ($label)').run(() async {
      final source = CsvSource(path: tempFile.path);
      return MemoryOptimizer.processInChunks(
        source.readLines(),
        const CsvOrderParser(),
        chunkSize: 50000,
      );
    });

    // Verificar se os resultados coincidem
    final streamRev = streamResult.revenue.toStringAsFixed(2);
    final chunkedRev = chunkedResult.revenue.toStringAsFixed(2);
    print('Receita coincide: ${streamRev == chunkedRev ? "PASS" : "FAIL"}');
    print('Receita: \$${streamRev} USD\n');

    await tempFile.delete();
  }

  // Benchmark 3: Verificação de tamanho do binário AOT
  print('--- Artefatos de Build ---');
  final exe = File('build/data_pipeline');
  if (await exe.exists()) {
    final sizeKB = await exe.length() ~/ 1024;
    print('Binário AOT: ${sizeKB}KB');
  } else {
    print('Binário AOT: não compilado (execute: dart compile exe bin/data_pipeline.dart)');
  }

  print('\n=== Benchmark Completo ===');
  print('Critérios de aceitação da Alice: 1M pedidos < 10s, memória < 300MB, binário < 20MB');
}

// ---- Stubs necessários (importados de lições anteriores) ----

sealed class DataSource {
  const DataSource();
  String get displayName;
  Stream<String> readLines();
}

class CsvSource extends DataSource {
  final String path;
  const CsvSource({required this.path});
  @override
  String get displayName => 'CSV: $path';
  @override
  Stream<String> readLines() =>
    File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

abstract class OrderParser {
  const OrderParser();
  Order parse(String raw);
}

class CsvOrderParser extends OrderParser {
  const CsvOrderParser();
  @override
  Order parse(String raw) {
    final parts = raw.split(',');
    if (parts.length < 4) throw FormatException('Linha inválida');
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
      region: parts.length > 4 ? parts[4].trim() : 'US',
    );
  }
}

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;
  final String region;
  const Order({required this.id, required this.amount, required this.status, required this.category, this.region = 'US'});
  double get tax => amount * 0.08;
}

class AnalysisResult {
  final int totalOrders;
  final int completedOrders;
  final double revenue;
  final double tax;
  final double total;
  final Map<String, double> revenueByCategory;
  const AnalysisResult({required this.totalOrders, required this.completedOrders, required this.revenue, required this.tax, required this.total, required this.revenueByCategory});
}

class OrderAnalyzer {
  const OrderAnalyzer();
  AnalysisResult analyze(List<Order> orders) {
    final completed = orders.where((o) => o.status == 'completed').toList();
    final revenue = completed.fold<double>(0, (s, o) => s + o.amount);
    final tax = completed.fold<double>(0, (s, o) => s + o.tax);
    final byCategory = <String, double>{};
    for (final o in completed) {
      byCategory.update(o.category, (v) => v + o.amount, ifAbsent: () => o.amount);
    }
    return AnalysisResult(totalOrders: orders.length, completedOrders: completed.length, revenue: revenue, tax: tax, total: revenue + tax, revenueByCategory: byCategory);
  }
}

class Pipeline {
  final DataSource source;
  final OrderParser parser;
  const Pipeline({required this.source, required this.parser});
  Future<AnalysisResult> execute() async {
    final orders = <Order>[];
    await for (final line in source.readLines()) {
      try { orders.add(parser.parse(line)); } on FormatException { continue; }
    }
    return const OrderAnalyzer().analyze(orders);
  }
}

class MemoryOptimizer {
  static Future<AnalysisResult> processInChunks(Stream<String> lines, OrderParser parser, {int chunkSize = 50000}) async {
    var chunk = <Order>[];
    int totalOrders = 0, completedOrders = 0;
    double revenue = 0, tax = 0;
    final byCategory = <String, double>{};

    await for (final line in lines) {
      try { chunk.add(parser.parse(line)); } on FormatException { continue; }
      if (chunk.length >= chunkSize) {
        final r = const OrderAnalyzer().analyze(chunk);
        totalOrders += r.totalOrders;
        completedOrders += r.completedOrders;
        revenue += r.revenue;
        tax += r.tax;
        for (final e in r.revenueByCategory.entries) {
          byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
        }
        chunk = [];
      }
    }
    if (chunk.isNotEmpty) {
      final r = const OrderAnalyzer().analyze(chunk);
      totalOrders += r.totalOrders;
      completedOrders += r.completedOrders;
      revenue += r.revenue;
      tax += r.tax;
      for (final e in r.revenueByCategory.entries) {
        byCategory.update(e.key, (v) => v + e.value, ifAbsent: () => e.value);
      }
    }
    return AnalysisResult(totalOrders: totalOrders, completedOrders: completedOrders, revenue: revenue, tax: tax, total: revenue + tax, revenueByCategory: byCategory);
  }
}

> **Saída:** Execute localmente no DartPad ou com `dart run`. Todos os exemplos do curso Dart são baseados em Dart 3.x / Flutter 3.x. Os resultados podem variar ligeiramente dependente da versão do SDK.