Add queue operator

[AlexandrHoroshih]

Queue

queue operator to setup queue of effects

Live example:
https://share.effector.dev/6fS9dyqn

const baseFx = createEffect(async p => {
  await new Promise(r => {
    setTimeout(r, 10 * p)
  })

  return p
})

const queuedFx = queue({effect: baseFx})

queuedFx(5)
queuedFx(4)
// next calls will not actually start, until previous effect is ended
// each queuedFx call returns Promise that resolves when this call is out of queue
// and resolved (with params and result of baseFx)
queuedFx(3)
queuedFx(2)
queuedFx(1)

// results will be in order of calls: 5, 4, 3, 2, 1
// without queue it would be 1, 2, 3, 4, 5, because baseFx in this example
// takes longer for bigger number in params

// calling baseFx without queue wrapper wont affect the queue
queuedFx.inFlight // value of this store will be equal to count of effects in the queue

Possible api, if there is a need of more control over a queue:
https://share.effector.dev/kPeWAIRZ

const cancel = createEvent()
const queuedFx = queue({ effect: baseFx, getKey: (params) => params.id, signal: cancel })

cancel("someId") // removes queued call with this key from the queue

in possible combination with abort

const cancel = createEvent()

const baseFx = abort({
  getKey: (p) => p.id,
  signal: cancel,
  handler: (p, {onAbort}) => { ... }
})

const fx = queue({
  signal: cancel,
  getKey: (p) => p.id,
  effect: baseFx
})

cancel("someId") // call with this key will be removed from queue if not in flight yet, if in flight - then cleaned up from queue and also aborted

UPD:

Updated snippet after meeting the real-world usecase:
https://share.effector.dev/7cmyUIyT

It turns out that we might need few separate queues + many different effects for each queue, so this api consists of two steps:

1. Creating the queue
2. Attaching effects to created queue

// 1
const myQueue = createQueue()

// 2
const qFx = myQueue({effect: fxSum1})
const qFx2 = myQueue({effect: fxSum2})

// or

const {qFx, qFx2} = myQueue({effect: {qFx: fx, qFx2: fx2}})

where each effect works described earlier

[igorkamyshev]

I think, it's better to start with a cancellable (abort) #154

What so you think?

[AlexandrHoroshih]

It is already in progress
https://github.com/AlexandrHoroshih/patronum/tree/feat/abort
and implementation is done - only problem is typescript typings are not good enough yet (require too much manual annotation)

[igorkamyshev]

Nice 😊

[doasync]

I suggest you first discuss the problem you are solving with this method.

[doasync]

@AlexandrHoroshih Do you have an actual use case?

[zerobias]

combination with abort

I suggest a shorthand for this:

const fx = queue({
  getKey: (p) => p.id,
  signal: cancel,
  async effect(p, {onAbort}) {
    
  }
})

[fnpen]

Hey everyone,

I am sharing my tests with queues and concurrency support:
It uses source and passes it to effect using attach,

https://share.effector.dev/EIThdjrq

Initialization:

const queue = createQueue({ concurrency: 4 });
const fx = createEffect(async ({ params: { time }, signal }) => {
    await new Promise((r, e) => {
      setTimeout(r, 3000);
    });
    
    console.log(`Finished: ${n}`, time);
    
    return time;
});

const $time = createStore(1000);

const abort = queue({
    source: $time,
    fn: (time) => ({ time }),
    effect: fx
});

Abort & Retry:

const fx = createEffect(async ({ params: { time }, signal }) => {
    await new Promise((r, e) => {
      signal.onabort = () => {
        e(`Aborted`);
      };
      setTimeout(r, 3000);
    });

    console.log(`Finished: ${n}`, time);

    return time;
  });

  fx.fail.watch(({ params, error }) => {
    console.error(`Aborted: ${n}`, error, params);

    if (params.retry) {
      params.retry();
    }
  });

  const abort = queue({
    source: $var,
    fn: (time) => ({ time }),
    effect: fx
  });

  if (n % 4 === 0 && n > 0) {
    fx.watch(({ retryAttempts }) => {
      if (retryAttempts < 2) {
        setTimeout(() => {
          abort();
        }, 1500);
      }
    });
 }

[sergeysova]

What about a queue length? To make N-parallel requests

[fnpen]

Yes, I tested with the big queue. The new version is much more effective, and you can make N-parallel requests, please check it

[fnpen]

I found performance issues in the previous example,
I decided to rewrite and restructure it.

Summary

• I removed a couple of unnecessary createEffect + new Promise
• AbortController one per queue - also speeds up and reduces memory usage
• 5x less memory waste - from 1.5 GB per 10,000 it is now 400 MB
• Faster addition - no need to run an effect during addition
• Every effect can be subscribed to done/watch from the queue
• Fewer CPUs at runtime
• Support for dynamic via source and params + fn
• Parallelism in the queue
• Priority in the queue via priority

Example: https://share.effector.dev/OPrqooUV

queue() returns event:

sample({
  clock: addToQueue,
  fn: () => ({ n: "text" }),
  target: queue({
    source: $time,
    fn: (time) => ({ time }),
    target: fetchDataFx,
  }),
});

forward({
  from: addToQueue,
  to: queue({
    source: $time,
    fn: (time) => ({ time }),
    target: fetchDataFx,
  }),
});

Abort:

const abort = createEvent();

const add = queue({
	priority: 1,
	target: actionFx,
	abort,
});

add();

sample({
	clock: cancel,
	filter: () => true,
	target: abort,
});

[fnpen]

New API, I decided to remove source/params/fn from the queue and take priority from the parameters,
I do not need to use any additional effect to add to the queue. Possible using samples to add effect to the queue and use event triggered in UI to abort the task in the queue.

it's possible to use it together with the sample:

const queue = createQueue({ concurrency: 4 });

const add = queue({
	target: taskFx,
});

add({ priority: 1, paramA: 1 });

or

const addAllTasksToQueue = createEvent();

sample({
    clock: addAllTasksToQueue,
    source: $time,
    fn: (time) => ({
        time,
        priority: 1,
    }),
    target: queue({
        target: taskFx,
    }),
});

addAllTasksToQueue();

Also possible to abort different effects using one abort event and match it using filterMap:

sample({
	clock: addAllTasksToQueue,
	source: $time,
	fn: (time) => ({
		time,
		n,
		priority: n,
	}),
	target: queue({
		target: taskFx,
		abort: clickRowIndex,
		abortFilter: (index, {n}) => index === n // Compares clickRowIndex event args with fx params
	}),
});

setTimeout(() => abortByIndex(n), 500);

Playground: https://share.effector.dev/qwdhnJpQ