Dart vs Kotlin: asynchronous generation

asynchronous_generation.dart

import 'dart:async';

void main() async {
  // Stream from Iterable
  final numList = [1, 2, 3];
  final numStream = Stream.fromIterable(numList);
  await for (final num in numStream) {
    print(num);
  }
  // Transforming Stream
  getNumbersStream()
      .where((num) => num.isEven)
      .map((evenNum) => "Even: $evenNum")
      .take(3)
      .handleError((e) => print("Error: $e"))
      .forEach((res) => print(res));
  // StreamController
  final eventBus = EventBus<String>();
  eventBus.events.forEach(print);
  eventBus.produceEvent("A");
  eventBus.produceEvent("B");
}

Stream<int> getNumbersStream() async* {
  var currentNumber = 0;
  while (true) {
    yield currentNumber++;
    await Future.delayed(Duration(milliseconds: 200));
  }
}

class EventBus<T> {
  StreamController<T> _controller = StreamController.broadcast();
  Stream<T> get events => _controller.stream;

  void produceEvent(T event) {
    _controller.add(event);
  }
}

asynchronous_generation.kt

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

 fun main() = runBlocking {
    // Stream from Iterable
    val numList = listOf(1, 2, 3)
    val numFlow = numList.asFlow()
    numFlow.collect { num ->
        println(num)
    }
    // Transforming Stream
    val job1 = getNumbersStream()
            .filter { num -> num % 2 == 0 }
            .map { evenNum -> "Even: $evenNum" }
            .take(3)
            .catch { e -> println("Error: $e") }
            .onEach { res -> println(res) }
            .launchIn(this)
    // SharedFlow
    val eventBus = EventBus<String>()
    val job2 = eventBus.events.onEach(::println).launchIn(this)
    val job3 = launch {
        eventBus.produceEvent("A")
        eventBus.produceEvent("B")
    }
    job1.join()
    job3.join()
    job2.cancelAndJoin()
}

fun getNumbersStream(): Flow<Int> = flow {
    var currentNumber = 0
    while (true) {
        emit(currentNumber++)
        delay(200)
    }
}

class EventBus<T> {
    private val _events = MutableSharedFlow<T>()
    val events = _events.asSharedFlow()

    suspend fun produceEvent(event: T) {
        _events.emit(event)
    }
}