dart-lang / pool Goto Github PK

import 'dart:async';
import 'dart:collection';
import 'dart:mirrors';

import 'package:async/async.dart';
import 'package:pool/pool.dart';

void main() async {
  Queue<String> queue = Queue();
  Pool workerPool = new Pool(2, timeout: Duration(seconds: 1));
  queue.addAll(["Hey", "there", "dart", "team"]);

  // If the loop is "inactive" for 5 seconds, processing should stop.
  var inactivityTimer = RestartableTimer(Duration(seconds: 5), () {});

  while (true) {
    if (!inactivityTimer.isActive) {
      // RELEVANT LINE: The queue is empty, the inactivity timer has been triggered, there is currently no way besides reflection to check if the pool resources has all completed 
      // Wait until all all work is done
      if (workerPool.allocatedResources == 0) {
        print("All workers are done");
        break;
      }
    }

    // The queue is empty, wait to check again
    if (queue.isEmpty) {
      await Future.delayed(Duration(seconds: 5));
    } else {
     // We have work to do, create a new pooled resource to work on it
      workerPool.withResource(() async {
        await Future.delayed(Duration(seconds: 10));
        print(queue.removeFirst())
      });
      inactivityTimer.reset();
    }

    // If we pummel the workerPool with resource requests too fast, it locks up.
    await Future.delayed(Duration(milliseconds: 10));
  }
}

final result = callback();
if (result is Future<T>) {
  return await result;
} else {
  return result;
}

import 'package:pool/pool.dart';

main() async {
  final pool = Pool(2, timeout: Duration(seconds: 30));

  final List<Future> futures = [];

  try {
    for (var i = 0; i < 10; i++) {
      futures.add(
        pool.withResource(
          () => Future.delayed(
            Duration(seconds: 1),
            () {
              if (i == 5) {
                throw 'Oh no!';
              }
              print(i);
              return i;
            },
          ),
        ),
      );
    }

    await Future.wait(futures, eagerError: true);
  } catch (e) {
    print('CAUGHT: $e');
  }
}

import 'package:pool/pool.dart';

main() async {
  bool stopSignal = false;

  final pool = Pool(2, timeout: Duration(seconds: 30));

  final List<Future> futures = [];

  try {
    for (var i = 0; i < 10; i++) {
      futures.add(
        pool.withResource(
          () => stopSignal == true
              ? Future(() => null)
              : Future.delayed(
                  Duration(seconds: 1),
                  () {
                    if (i == 5) {
                      throw 'Oh no!';
                    }
                    print(i);
                    return i;
                  },
                ),
        ),
      );
    }

    await Future.wait(futures, eagerError: true);

  } catch (e) {
    print('CAUGHT: $e');
    stopSignal = true;
  }
}

Stream<T> pooledIterator<S, T>(
    int poolSize, Iterable<S> sourceItems, Future<T> action(S source),
    {bool swallowErrors(S item, Object error, StackTrace stack)}) async* {
  swallowErrors ??= (item, e, s) => false;
  final pool = Pool(poolSize);
  final controller = StreamController<T>();

  final items = Future.wait(sourceItems.map((item) async {
    final resource = await pool.request();

    try {
      controller.add(await action(item));
    } catch (e, stack) {
      if (!swallowErrors(item, e, stack)) {
        controller.addError(e, stack);
      }
    } finally {
      resource.release();
    }
  })).whenComplete(() async {
    await controller.close();
  });

  yield* controller.stream;

  await items;
}

R Function(P1, P2) limit2<P1, P2>(R Function(P1, P2) callback) =>
  (p1, p2) => withResource(() => callback(p1, p2));

import 'dart:async';
import 'package:pool/pool.dart';

main() async {
  Pool pool = new Pool(3);

  for (int i = 0; i < 4; i++) {
    pool.withResource(() {
      print(i);
    });
  }

  await pool.close();
}