construct stateful functions that can be called any number of times

concurrency-constraining-functions.js

import { execute, timeout, call } from 'effection';


// Rather than shooting for some grandiose templating ability for
// concurrency (I still think we can get there, I just don't see how
// at the moment).  What if we take a smaller step: Given some
// generator, how can we use it to generate a single function that can
// be called over and over and that internally it will manage a set of
// tasks instances for us?
//
// If we can find out what it takes to generate this function, then
// maybe we can figure out how to compose them into a greater tree.
//
// Take for example this task that counts down from a number.

function* countdownFrom(start) {
  for (let i = start; i > 1; i--) {
    console.log(`${i}...`);
    yield timeout(1000);
  }
  console.log('liftoff!');
}

// Let's start with probably the most useful one: a function that
// constrains how many instances of a task can be running at a time.
// throttle() will return a function that can be called again and
// again, but it will track concurrency, and ignore calls that would
// cause concurrency to exceed the maximum.

export function throttle(proc, maxConcurrency) {
  let concurrency = 0;

  return function(...args) {
    if (concurrency < maxConcurrency) {
      concurrency++;
      execute(function*() {
        try {
          yield call(proc, ...args);
        } finally {
          concurrency--;
        }
      });
    }
  };
}

// We can now use this function to create a throttled task:

let countdown = throttle(countdownFrom, 2);

countdown(5); //=> starts a countdown from 5
countdown(10); //=> starts a coundown from 10
countdown(2); //=> ignored!
countdown(2); //=> ignored!

// 5s pass
countdown(2); //=> starts a coundown from 2

// We can do the same for `restart`, `drop`, and `enqueue`. In each
// case, we're returning a function that can be called repeatedly
// and the task.

export function restart(proc) {
  let current = { halt() {} };

  return function(...args) {
    current.halt();
    current = execute(proc, ...args);
  };
}

export function drop(proc) {
  let current = { isPending: false };

  return function(...args) {
    if (!current.isPending) {
      current = execute(proc, ...args);
    }
  };
}

// enqueue is like the linked list of task concurrency.
// it starts out with an empty generator as the tail, and every time
// you enqueue a set of arguments, it forks a new task
// that yields to tail, and then calls the new one.
// that forked taks now becomes the new tail.
// recursion ftw.
export function enqueue(proc) {
  let loop = execute(function*() {
    let tail = function*() {};

    while (true) {
      let args = yield;
      tail = this.fork(function*(prior) {
        yield prior;
        yield call(proc, ...args);
      }, [tail]);
    }
  });

  return function(...args) {
    loop.resume(args);
  };
}


// Using this technique, we can build our guard,
// that only runs a task when the condition is met
// for the first time.
export function when(condition, proc) {
  let isMatching = false;
  let current = { halt() {} };
  return function(...args) {
    if (condition(...args)) {
      if (isMatching) {
        // do nothing, we're already matching.
      } else {
        current = execute(proc, ...args);
      }
    } else {
      if (isMatching) {
        current.halt();
      }
    }
  };
}

let maybeCountdown = when(count => count < 10, countdownFrom);

maybeCountdown(20); //=> two high, does nothing.

maybeCountdown(5); //=> works, because predicate passed.
maybeCountdown(5); //=> does nathing, because condition is currently matching.
maybeCountdown(20); //=> cancels current countdown because match failed.