Dart: Stream في Dart

orderStream
  .where((order) => order.status == 'completed')
  .map((order) => order.amount)
  .fold(0, (sum, amount) => sum + amount)
  .then((total) => print('Revenue: \$$total USD'));

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  // إنشاء stream من iterable
  final stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  // الاستماع إلى stream
  await for (final value in stream) {
    print('Received: $value');
  }

  // بديل: الاستماع مع استدعاء
  final stream2 = Stream.fromIterable(['ORD-001', 'ORD-002', 'ORD-003']);
  stream2.listen(
    (order) => print('Order: $order'),
    onDone: () => print('Stream completed'),
    onError: (e) => print('Error: $e'),
  );

  // انتظار اكتمال stream
  await stream2.drain();
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  // من iterable - يصدر كل عنصر
  final orders = Stream.fromIterable([
    {'id': 'ORD-001', 'amount': 1500.0},
    {'id': 'ORD-002', 'amount': 3200.0},
    {'id': 'ORD-003', 'amount': 890.0},
  ]);

  await for (final order in orders) {
    print('Order: ${order['id']} - \$${order['amount']} USD');
  }
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  // Stream دوري - يصدر قيماً على فترات منتظمة
  final ticker = Stream.periodic(
    const Duration(seconds: 1),
    (count) => 'Tick ${count + 1}',
  );

  // خذ أول 5 علامات فقط
  await for (final tick in ticker.take(5)) {
    print(tick);
  }
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  final controller = StreamController<String>();

  // إضافة بيانات إلى الـ stream
  controller.add('ORD-001');
  controller.add('ORD-002');
  controller.add('ORD-003');

  // إغلاق الـ stream عند الانتهاء
  controller.close();

  // استهلاك الـ stream
  await for (final order in controller.stream) {
    print('Processing: $order');
  }
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  final amounts = Stream.fromIterable([1500.0, 50.0, 3200.0, 890.0, -100.0]);

  // تصفية المبالغ الموجبة فقط >= 100
  final highValue = amounts.where((a) => a >= 100);

  await for (final amount in highValue) {
    print('\$${amount.toStringAsFixed(2)} USD');
  }
  // الإخراج: $1500.00 USD, $3200.00 USD, $890.00 USD
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  final amounts = Stream.fromIterable([1500.0, 3200.0, 890.0]);

  // تحويل المبالغ إلى سلاسل منسقة
  final formatted = amounts.map((a) => '\$${a.toStringAsFixed(2)} USD');

  await for (final text in formatted) {
    print(text);
  }
  // الإخراج: $1500.00 USD, $3200.00 USD, $890.00 USD
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  final orders = Stream.fromIterable([
    {'id': 'ORD-001', 'items': ['Laptop', 'Mouse']},
    {'id': 'ORD-002', 'items': ['Keyboard']},
  ]);

  // expand: كل طلب ← عناصر متعددة
  final items = orders.expand((order) =>
      (order['items'] as List).cast<String>());

  await for (final item in items) {
    print('Item: $item');
  }
  // الإخراج: Item: Laptop, Item: Mouse, Item: Keyboard
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  final data = Stream.fromIterable([1, 1, 2, 2, 3, 3, 4, 5]);

  // خذ أول 3 عناصر
  print('take(3):');
  await data.take(3).forEach(print);  // 1, 1, 2

  // تخطي أول 2
  print('skip(2):');
  await Stream.fromIterable([1, 1, 2, 2, 3]).skip(2).forEach(print);  // 2, 2, 3

  // إزالة التكرارات المتتالية
  print('distinct:');
  await Stream.fromIterable([1, 1, 2, 2, 3, 3]).distinct().forEach(print);  // 1, 2, 3
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

class OrderStream {
  final _controller = StreamController<Map<String, dynamic>>();

  Stream<Map<String, dynamic>> get stream => _controller.stream;

  void addOrder(Map<String, dynamic> order) => _controller.add(order);

  void addError(Object error) => _controller.addError(error);

  void close() => _controller.close();
}

void main() async {
  final orderStream = OrderStream();

  // الاشتراك في الـ stream
  final subscription = orderStream.stream.listen(
    (order) => print('Processing: ${order['id']}'),
    onError: (e) => print('Error: $e'),
    onDone: () => print('Stream closed'),
  );

  // إضافة الطلبات
  orderStream.addOrder({'id': 'ORD-001', 'amount': 1500.0});
  orderStream.addOrder({'id': 'ORD-002', 'amount': 3200.0});
  orderStream.addOrder({'id': 'ORD-003', 'amount': 890.0});

  // إغلاق الـ stream
  orderStream.close();

  // انتظار انتهاء الـ stream
  await subscription.asFuture();
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

void main() async {
  // controller بث - مستمعون متعددون
  final controller = StreamController<String>.broadcast();

  // مشتركون متعددون
  final sub1 = controller.stream.listen((e) => print('Sub1: $e'));
  final sub2 = controller.stream.listen((e) => print('Sub2: $e'));

  // كلا المستمعين يتلقيان نفس الحدث
  controller.add('ORD-001');
  controller.add('ORD-002');

  controller.close();
  await Future.delayed(const Duration(milliseconds: 100));
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

import 'dart:async';

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;

  Order({required this.id, required this.amount, required this.status, required this.category});

  @override
  String toString() => 'Order($id, \$${amount.toStringAsFixed(2)}, $status, $category)';
}

void main() async {
  // محاكاة stream الطلبات في الوقت الفعلي
  final controller = StreamController<Order>();

  // بناء خط أنابيب المعالجة
  final revenueByCategory = <String, double>{};
  var totalOrders = 0;
  var completedOrders = 0;

  controller.stream
      .where((order) => order.amount > 0)       // تصفية غير الصالح
      .where((order) => order.status == 'completed')  // المكتملة فقط
      .map((order) => order)                     // يمكن التحويل هنا
      .listen(
        (order) {
          completedOrders++;
          totalOrders++;
          revenueByCategory.update(
            order.category,
            (v) => v + order.amount,
            ifAbsent: () => order.amount,
          );
          print('  Processed: ${order.id} - \$${order.amount.toStringAsFixed(2)} USD');
        },
        onDone: () {
          print('\n=== Stream Processing Complete ===');
          print('Completed orders: $completedOrders');
          for (final entry in revenueByCategory.entries) {
            print('  ${entry.key}: \$${entry.value.toStringAsFixed(2)} USD');
          }
          final total = revenueByCategory.values.fold(0.0, (a, b) => a + b);
          print('Total revenue: \$${total.toStringAsFixed(2)} USD');
        },
      );

  // محاكاة الطلبات الواردة
  controller.add(Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'));
  controller.add(Order(id: 'ORD-002', amount: -50.0, status: 'completed', category: 'Books'));      // غير صالح
  controller.add(Order(id: 'ORD-003', amount: 3200.0, status: 'pending', category: 'Electronics'));  // ليس مكتمل
  controller.add(Order(id: 'ORD-004', amount: 890.0, status: 'completed', category: 'Clothing'));
  controller.add(Order(id: 'ORD-005', amount: 2100.0, status: 'completed', category: 'Electronics'));

  controller.close();
  await Future.delayed(const Duration(milliseconds: 100));
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

// ============================================
// خط أنابيب معالجة Stream لخط أنابيب البيانات
// stream طلبات في الوقت الفعلي مع عوامل
// ============================================

import 'dart:async';

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 taxAmount => amount * 0.08;
  double get totalWithTax => amount + taxAmount;

  @override
  String toString() =>
      'Order($id, \$${amount.toStringAsFixed(2)}, $status, $category, $region)';
}

class StreamPipeline {
  final _controller = StreamController<Order>();
  final Map<String, double> _revenue = {};
  final Map<String, int> _count = {};
  int _totalProcessed = 0;
  int _totalSkipped = 0;
  late StreamSubscription<Order> _subscription;

  Stream<Order> get stream => _controller.stream;

  StreamPipeline() {
    _subscription = _controller.stream
        .where((o) => o.amount > 0)
        .where((o) => o.status != 'cancelled')
        .distinct((o) => o.id)  // إزالة التكرار حسب المعرف
        .listen(
          _processOrder,
          onError: (e) => print('Pipeline error: $e'),
          onDone: _printSummary,
        );
  }

  void _processOrder(Order order) {
    _totalProcessed++;
    _revenue.update(
      order.category,
      (v) => v + order.totalWithTax,
      ifAbsent: () => order.totalWithTax,
    );
    _count.update(order.category, (v) => v + 1, ifAbsent: () => 1);
  }

  void addOrder(Order order) => _controller.add(order);

  void addError(Object error) => _controller.addError(error);

  Future<void> close() async {
    await _controller.close();
    await _subscription.asFuture();
  }

  void _printSummary() {
    final totalRevenue = _revenue.values.fold(0.0, (a, b) => a + b);
    final totalCount = _count.values.fold(0, (a, b) => a + b);

    print('\n=== DataPipeline Stream Report ===');
    print('Processed: $_totalProcessed orders');
    print('Skipped:   $_totalSkipped records');
    print('Revenue:   \$${totalRevenue.toStringAsFixed(2)} USD');
    print('Orders:    $totalCount');

    print('\n--- By Category ---');
    final sorted = _revenue.entries.toList()
      ..sort((a, b) => b.value.compareTo(a.value));
    for (final entry in sorted) {
      final count = _count[entry.key] ?? 0;
      final avg = entry.value / count;
      print('  ${entry.key}:');
      print('    Orders:  $count');
      print('    Revenue: \$${entry.value.toStringAsFixed(2)} USD');
      print('    Average: \$${avg.toStringAsFixed(2)} USD');
    }
  }
}

void main() async {
  final pipeline = StreamPipeline();

  // محاكاة stream الطلبات في الوقت الفعلي
  final orders = [
    Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'),
    Order(id: 'ORD-002', amount: -50.0, status: 'completed', category: 'Books'),
    Order(id: 'ORD-003', amount: 3200.0, status: 'pending', category: 'Electronics'),
    Order(id: 'ORD-004', amount: 890.0, status: 'completed', category: 'Clothing'),
    Order(id: 'ORD-001', amount: 1500.0, status: 'completed', category: 'Electronics'),  // مكرر
    Order(id: 'ORD-005', amount: 2100.0, status: 'completed', category: 'Electronics'),
    Order(id: 'ORD-006', amount: 500.0, status: 'cancelled', category: 'Books'),
    Order(id: 'ORD-007', amount: 120.0, status: 'completed', category: 'Books'),
  ];

  // إصدار الطلبات واحداً تلو الآخر (محاكاة الوقت الفعلي)
  for (final order in orders) {
    pipeline.addOrder(order);
  }

  await pipeline.close();
}

> **الإخراج:** شغّل محلياً في DartPad أو عبر `dart run`. جميع الأمثلة في دورة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً حسب إصدار SDK.

=== DataPipeline Stream Report ===
Processed: 5 orders
Skipped:   0 records
Revenue:   $8154.00 USD
Orders:    5

--- By Category ---
  Electronics:
    Orders:  3
    Revenue: $7236.00 USD
    Average: $2412.00 USD
  Clothing:
    Orders:  1
    Revenue: $961.20 USD
    Average: $961.20 USD
  Books:
    Orders:  1
    Revenue: $129.60 USD
    Average: $129.60 USD