discontinuedPackage connect_database has been discontinued.

streamBuffered method

Future<Result<Stream<List<Map<String, dynamic>>>>> streamBuffered({ int bufferSize = 100, })

Streams data in batches to balance memory usage and performance. Ideal for very large datasets where you want to control memory footprint. bufferSize - Number of rows to fetch per batch (default: 100)

Example:

final result = await cmd.streamBuffered(bufferSize: 500);
await for (final batch in result.getOrThrow()) {
  for (final row in batch) {
    // Process each row
  }
}

Implementation

Future<Result<Stream<List<Map<String, dynamic>>>>> streamBuffered({
  int bufferSize = 100,
}) async {
  try {
    if (!_isConnected) {
      await odbc.connect().getOrThrow();
      _isConnected = true;
    }

    var query = commandText;
    if (query == null || query.isEmpty) {
      return Failure(Exception('CommandText não pode ser vazio.'));
    }

    if (_useReadUncommitted) {
      if (odbc.type == DatabaseType.sqlServer ||
          odbc.type == DatabaseType.sybaseAnywhere) {
        query = _addNoLock(query);
      }
    }

    if (SqlValidCommand.isSelectQuery(query)) {
      if (!_disableInterceptorForStream) {
        final interceptedResult =
            await _selectInterceptor.interceptSelect(query);
        if (interceptedResult.isError()) {
          return Failure(interceptedResult.exceptionOrNull()!);
        }
        query = interceptedResult.getOrThrow();
      }
    }

    final preparedResult = _substituteParameters(query);
    if (preparedResult.isError()) {
      return Failure(preparedResult.exceptionOrNull()!);
    }
    final prepared = preparedResult.getOrThrow();

    // Get cursor stream
    final cursorResult =
        await odbc.executeCursor(prepared.sql, params: prepared.params);
    if (cursorResult.isError()) {
      return Failure(cursorResult.exceptionOrNull()!);
    }

    final rowStream = cursorResult.getOrThrow();

    // Convert to buffered batch stream
    Stream<List<Map<String, dynamic>>> bufferedStream() async* {
      final buffer = <Map<String, dynamic>>[];

      await for (final row in rowStream) {
        buffer.add(row);

        if (buffer.length >= bufferSize) {
          yield List<Map<String, dynamic>>.from(buffer);
          buffer.clear();
        }
      }

      // Yield remaining rows
      if (buffer.isNotEmpty) {
        yield buffer;
      }
    }

    return Success(bufferedStream());
  } catch (e, s) {
    return Failure(Exception('Error opening buffered stream: $e\nStack: $s'));
  }
}