Isolate #

import 'dart:isolate';

Future<void> main() async {
  // Create a ReceivePort — an endpoint for receiving messages
  final receivePort = ReceivePort();

  // sendPort is the "address" that can be sent to another Isolate
  final SendPort sendPort = receivePort.sendPort;

  // Send a message to ourselves (for demonstration)
  sendPort.send('Halo!');
  sendPort.send(42);
  sendPort.send({'kunci': 'nilai'});
  sendPort.send(null);

  // Read messages — a ReceivePort is a Stream
  int hitungan = 0;
  await for (final pesan in receivePort) {
    print('Received: $pesan');
    hitungan++;
    if (hitungan >= 4) break; // stop after 4 messages
  }

  receivePort.close(); // must close to stop the event loop
}

import 'dart:isolate';
import 'dart:typed_data';

final port = ReceivePort();
final send = port.sendPort;

// ✓ CAN be sent between Isolates:
send.send(null);                          // null
send.send(true);                          // bool
send.send(42);                            // int
send.send(3.14);                          // double
send.send('teks');                        // String
send.send([1, 2, 3]);                     // List (copied)
send.send({'a': 1});                      // Map (copied)
send.send({1, 2, 3});                     // Set (copied)
send.send(Uint8List.fromList([1, 2, 3])); // TypedData (copied)
send.send(port.sendPort);                 // SendPort ✓
send.send(                                // TransferableTypedData (zero-copy)
  TransferableTypedData.fromList([Uint8List(100)])
);

// ✗ CANNOT be sent:
// send.send(File('test.txt'));    // native resource objects
// send.send(Socket(...));         // native sockets
// send.send(() => print('hi'));   // closures/functions (except top-level)
// send.send(Stream.empty());      // Streams

import 'dart:isolate';

// Isolate.run — create, run, return the result, clean up automatically
Future<void> main() async {
  // The function run in a new Isolate
  // Must be top-level or static (can't be a closure that captures state)
  final hasil = await Isolate.run(() {
    // This code runs in a separate Isolate
    int total = 0;
    for (int i = 0; i < 10000000; i++) {
      total += i;
    }
    return total;
  });

  print('Total: $hasil'); // 49999995000000
}

// With arguments — send data via the closure
Future<List<int>> sortkanDiBackground(List<int> data) async {
  return await Isolate.run(() {
    // data is copied into this Isolate
    final salinan = List<int>.from(data);
    salinan.sort();
    return salinan;
  });
}

// Errors in Isolate.run — propagate to the caller
Future<void> contohError() async {
  try {
    await Isolate.run(() {
      throw Exception('Error in the Isolate!');
    });
  } catch (e) {
    print('Error caught in main: $e');
    // Exceptions from Isolate.run are wrapped as RemoteError
  }
}

import 'dart:isolate';

// The entry point MUST be a top-level or static function
void workerEntryPoint(SendPort kirimKeMain) {
  final terimaPerintah = ReceivePort();

  // Send our SendPort to main so main can send commands
  kirimKeMain.send(terimaPerintah.sendPort);

  // Process incoming commands
  terimaPerintah.listen((pesan) {
    if (pesan == null) {
      // Signal to close
      terimaPerintah.close();
      return;
    }

    final int angka = pesan as int;
    final hasil = _hitungPrime(angka); // heavy computation
    kirimKeMain.send(hasil);
  });
}

bool _hitungPrime(int n) {
  if (n < 2) return false;
  for (int i = 2; i <= n ~/ 2; i++) {
    if (n % i == 0) return false;
  }
  return true;
}

Future<void> main() async {
  final terimaMain = ReceivePort();

  // Spawn the worker Isolate
  final isolate = await Isolate.spawn(
    workerEntryPoint,
    terimaMain.sendPort,
  );

  // Receive the SendPort from the worker
  final SendPort kirimKeWorker = await terimaMain.first;

  // Send several jobs
  final hasilStream = terimaMain.skip(1); // skip the SendPort already taken
  final iterator = hasilStream.iterator;

  for (final n in [97, 100, 101, 200, 997]) {
    kirimKeWorker.send(n);
    await iterator.moveNext();
    final isPrime = iterator.current as bool;
    print('$n is a prime number: $isPrime');
  }

  // Send the close signal
  kirimKeWorker.send(null);
  terimaMain.close();

  // Kill the Isolate if it's still alive
  isolate.kill(priority: Isolate.beforeNextEvent);
}

import 'dart:isolate';
import 'dart:async';

// Request message from main to worker
class Request {
  final int id;
  final dynamic data;
  final SendPort replyTo;

  const Request({required this.id, required this.data, required this.replyTo});
}

// Response message from worker to main
class Response {
  final int id;
  final dynamic hasil;
  final String? error;

  const Response({required this.id, this.hasil, this.error});
}

// A reliable worker with ID tracking
void workerRobust(SendPort kirimKeMain) {
  final terima = ReceivePort();
  kirimKeMain.send(terima.sendPort);

  terima.listen((pesan) {
    if (pesan is Request) {
      try {
        final hasil = _prosesData(pesan.data);
        pesan.replyTo.send(Response(id: pesan.id, hasil: hasil));
      } catch (e) {
        pesan.replyTo.send(Response(id: pesan.id, error: e.toString()));
      }
    }
  });
}

dynamic _prosesData(dynamic data) {
  // Simulated computation
  return 'Hasil dari: $data';
}

// A thread-safe client
class IsolateClient {
  final SendPort _kirimKeWorker;
  final ReceivePort _terima;
  final Map<int, Completer<dynamic>> _pending = {};
  int _idCounter = 0;

  IsolateClient._(this._kirimKeWorker, this._terima) {
    _terima.listen((pesan) {
      if (pesan is Response) {
        final completer = _pending.remove(pesan.id);
        if (pesan.error != null) {
          completer?.completeError(Exception(pesan.error));
        } else {
          completer?.complete(pesan.hasil);
        }
      }
    });
  }

  static Future<IsolateClient> buat() async {
    final terima = ReceivePort();
    await Isolate.spawn(workerRobust, terima.sendPort);
    final SendPort kirimKeWorker = await terima.first;
    return IsolateClient._(kirimKeWorker, terima);
  }

  Future<dynamic> kirim(dynamic data, {Duration timeout = const Duration(seconds: 30)}) {
    final id = _idCounter++;
    final completer = Completer<dynamic>();
    _pending[id] = completer;

    _kirimKeWorker.send(Request(
      id: id,
      data: data,
      replyTo: _terima.sendPort,
    ));

    return completer.future.timeout(
      timeout,
      onTimeout: () {
        _pending.remove(id);
        throw TimeoutException('Request $id timed out after ${timeout.inSeconds}s');
      },
    );
  }

  void tutup() {
    _terima.close();
    for (final c in _pending.values) {
      c.completeError(StateError('Client closed'));
    }
    _pending.clear();
  }
}

// Usage
Future<void> main() async {
  final client = await IsolateClient.buat();

  // Send many requests in parallel
  final futures = [
    client.kirim('data A'),
    client.kirim('data B'),
    client.kirim('data C'),
  ];

  final hasil = await Future.wait(futures);
  print(hasil); // ['Hasil dari: data A', ...]

  client.tutup();
}

import 'dart:isolate';

Future<void> main() async {
  final terimaError = ReceivePort();
  final terimaExit = ReceivePort();
  final terimaHasil = ReceivePort();

  final isolate = await Isolate.spawn(
    workerDenganError,
    terimaHasil.sendPort,
    // Error listener — catch uncaught errors in the Isolate
    onError: terimaError.sendPort,
    // Exit listener — notification when the Isolate finishes
    onExit: terimaExit.sendPort,
    // Errors DON'T kill the Isolate (default: true = kills it)
    errorsAreFatal: false,
  );

  // Listen for errors
  terimaError.listen((error) {
    // error is a List: [errorMessage, stackTrace]
    final List<dynamic> errorList = error as List<dynamic>;
    print('Error in the Isolate: ${errorList[0]}');
    print('Stack trace: ${errorList[1]}');
  });

  // Listen for exit
  terimaExit.first.then((_) {
    print('Isolate finished');
    terimaError.close();
    terimaExit.close();
    terimaHasil.close();
  });

  // Listen for results
  terimaHasil.listen((pesan) {
    print('Result: $pesan');
  });

  // Control from outside
  // Add an error listener to an already running Isolate
  isolate.addErrorListener(terimaError.sendPort);
  isolate.removeErrorListener(terimaError.sendPort);

  // Pause and resume
  final capability = isolate.pause();
  await Future.delayed(Duration(seconds: 1));
  isolate.resume(capability);

  // Send a kill signal
  // isolate.kill(priority: Isolate.immediate);  // immediately
  // isolate.kill(priority: Isolate.beforeNextEvent); // after the current event
}

void workerDenganError(SendPort kirimHasil) {
  kirimHasil.send('Started working');

  try {
    kirimHasil.send('Step 1 done');
    throw Exception('Simulated error');
    kirimHasil.send('This is never reached');
  } catch (e) {
    kirimHasil.send('Error caught locally: $e');
  }

  // An uncaught error will be sent to the onError port
  throw StateError('Uncaught error!');
}

import 'dart:isolate';

Future<void> main() async {
  // RawReceivePort — no Stream overhead, direct callback
  final rawPort = RawReceivePort((pesan) {
    print('Received (raw): $pesan');
  });

  rawPort.sendPort.send('Halo');
  rawPort.sendPort.send(42);

  await Future.delayed(Duration(milliseconds: 100));

  // Change the handler
  rawPort.handler = (pesan) {
    print('New handler: $pesan');
  };

  rawPort.sendPort.send('New message');

  await Future.delayed(Duration(milliseconds: 100));
  rawPort.close(); // must close
}

import 'dart:isolate';
import 'dart:async';
import 'dart:collection';

typedef IsolateTask<T> = Future<T> Function();

class IsolatePool {
  final int ukuran;
  final _workers = <_WorkerIsolate>[];
  final _antrian = Queue<_PendingTask>();
  bool _ditutup = false;

  IsolatePool(this.ukuran);

  Future<void> inisialisasi() async {
    for (int i = 0; i < ukuran; i++) {
      final worker = await _WorkerIsolate.buat();
      _workers.add(worker);
    }
  }

  Future<T> jalankan<T>(dynamic Function() tugas) async {
    if (_ditutup) throw StateError('Pool is closed');

    final completer = Completer<T>();
    final pending = _PendingTask(tugas, completer as Completer<dynamic>);

    // Find a free worker
    final workerKosong = _workers.firstWhere(
      (w) => !w.sibuk,
      orElse: () => throw StateError('No worker available'),
    );

    if (!workerKosong.sibuk) {
      await workerKosong.jalankan(pending);
    } else {
      _antrian.addLast(pending);
    }

    return completer.future as Future<T>;
  }

  Future<void> tutup() async {
    _ditutup = true;
    for (final worker in _workers) {
      await worker.tutup();
    }
  }
}

class _PendingTask {
  final dynamic Function() tugas;
  final Completer<dynamic> completer;
  _PendingTask(this.tugas, this.completer);
}

class _WorkerIsolate {
  bool sibuk = false;
  late final SendPort _kirimKeWorker;
  late final ReceivePort _terima;
  late final Isolate _isolate;

  static Future<_WorkerIsolate> buat() async {
    final worker = _WorkerIsolate();
    worker._terima = ReceivePort();
    worker._isolate = await Isolate.spawn(
      _workerLoop,
      worker._terima.sendPort,
    );
    worker._kirimKeWorker = await worker._terima.first;
    return worker;
  }

  Future<void> jalankan(_PendingTask task) async {
    sibuk = true;
    _kirimKeWorker.send(task.tugas);
    final hasil = await _terima.first;
    task.completer.complete(hasil);
    sibuk = false;
  }

  Future<void> tutup() async {
    _kirimKeWorker.send(null); // close signal
    _terima.close();
    _isolate.kill();
  }
}

void _workerLoop(SendPort kirimKeMain) {
  final terima = ReceivePort();
  kirimKeMain.send(terima.sendPort);

  terima.listen((tugas) {
    if (tugas == null) {
      terima.close();
      return;
    }
    final hasil = (tugas as dynamic Function())();
    kirimKeMain.send(hasil);
  });
}

// Pool usage
Future<void> main() async {
  final pool = IsolatePool(4); // 4 worker Isolates
  await pool.inisialisasi();

  // Send many tasks in parallel
  final futures = List.generate(
    10,
    (i) => pool.jalankan<int>(() {
      // Simulated heavy computation
      int sum = 0;
      for (int j = 0; j < 1000000; j++) sum += j;
      return sum + i;
    }),
  );

  final hasil = await Future.wait(futures);
  print('Result: ${hasil.length} tasks completed');

  await pool.tutup();
}

import 'dart:isolate';

Future<void> main() async {
  final isolate = await Isolate.spawn(
    (SendPort _) { /* ... */ },
    ReceivePort().sendPort,
  );

  // Pause returns a Capability as the "key"
  final pauseCapability = isolate.pause();
  print('Isolate paused');

  await Future.delayed(Duration(seconds: 2));

  // Resume can only be done by whoever holds the Capability
  isolate.resume(pauseCapability);
  print('Isolate resumed');

  // Capabilities can also be created independently
  final cap1 = Capability();
  final cap2 = Capability();
  print(cap1 == cap2); // false — every Capability is unique
  print(cap1 == cap1); // true — exactly the same
}

// ANTI-PATTERN: ReceivePort not closed — the event loop can't stop!
Future<void> buruk() async {
  final rp = ReceivePort();
  await Isolate.spawn(worker, rp.sendPort);
  final hasil = await rp.first;
  // ✗ rp is not closed — the program will never finish!
  print(hasil);
}

// CORRECT: always close the ReceivePort when done
Future<void> baik() async {
  final rp = ReceivePort();
  try {
    await Isolate.spawn(worker, rp.sendPort);
    final hasil = await rp.first;
    print(hasil);
  } finally {
    rp.close(); // ✓ always closed
  }
}

// ANTI-PATTERN: sending non-serializable objects
class ProdukDenganCallback {
  final String nama;
  final void Function() onUpdate; // callback/closure

  ProdukDenganCallback(this.nama, this.onUpdate);
}

sendPort.send(ProdukDenganCallback('Laptop', () {}));
// ✗ Illegal argument in isolate message: closure

// CORRECT: send only serializable data
sendPort.send({'nama': 'Laptop', 'harga': 15000000}); // ✓ Maps can be sent