Как очистить очередь.zip?

Question

Я создал простую "систему игровых циклов", основанную на RxJS, где я запрашиваю действия у разных участников (включая пользователя) и объединяю их с оператором.zip.

Результатом является пошаговый игровой цикл, который приостанавливается только для анимации и для запроса действий пользователя.

Благодаря этой системе пользователь может ставить в очередь несколько действий, и система будет хранить их в.zip-операторе. Однако я хочу очистить эту "очередь", если пользователь решит передумать.

У меня есть скрипка: https://jsfiddle.net/az5pthng/6/

Представьте, что пользователь вводит последовательность a1, a2, a3, a4, затем думает секунду и решает изменить последовательность на b1, b2, b3, b4. Затем снова думает и набирает c1, c2, c3, c4.

В настоящее время у меня есть этот вывод, показывающий, что каждое нажатие клавиши записывается и воспроизводится:

(3) ["A", "B", "a1"]
(3) ["A", "B", "a2"]
1100
(3) ["A", "B", "a3"]
(3) ["A", "B", "a4"]
1723
(3) ["A", "B", "b1"]
(3) ["A", "B", "b2"]
(3) ["A", "B", "b3"]
(3) ["A", "B", "b4"]
(3) ["A", "B", "c1"]
(3) ["A", "B", "c2"]
(3) ["A", "B", "c3"]
(3) ["A", "B", "c4"]

Но я хочу этот вывод, указывающий, что всякий раз, когда queueTime$ выдает значение - queue$ получает "сброс" с новыми значениями:

(3) ["A", "B", "a1"]
(3) ["A", "B", "a2"]
1100
(3) ["A", "B", "b1"]
(3) ["A", "B", "b2"]
1700
(3) ["A", "B", "c1"]
(3) ["A", "B", "c2"]
(3) ["A", "B", "c3"]
(3) ["A", "B", "c4"]

РЕДАКТИРОВАТЬ: Вот картина, что я пытаюсь сделать:

Два потока - Turn и Q

Turn должен как-то запросить A и B и подождать, пока Q предоставит его значение, если оно есть. Когда Q предоставляет значение, Turn ждет некоторое время, а затем повторяется. Q, однако, разбивает его значения по времени (я думаю, с помощью.window?) И отбрасывает любое значение, которое не отправлено в Turn, если поступает новая последовательность.

Answer 1

Не уверен, правильно ли я понимаю ваш вопрос, но если вы хотите отложить очередь на запуск (поскольку пользователь может передумать, вы можете использовать debounceTime(1000), где 1000 в мс.
Если вы хотите очистить очередь, вы можете использовать тот же оператор с DifferentUntilChanged, чтобы распознать изменения и сделать то, что вы хотите! надеюсь, это поможет

Answer 2

Я это сделал! Хотя я чувствую, что это не так, как RxJS, но, по крайней мере, работает так, как я хочу. Если у вас есть лучшее решение - пожалуйста, напишите, я не очень рад иметь переменные и флаги "вне потока".

Мой способ - управлять очередью самостоятельно. А также есть флаг всякий раз, когда мне нужен ввод - если вся цепочка ожидает ввода, это правда, в противном случае нет. Основываясь на этом флаге, я либо помещаю нажатую клавишу в очередь, либо отправляю в цепочку напрямую.

input$.subscribe(({value, interval}) => {
    if (waitingForInput) {
    queue$.next(value);
    waitingForInput = false;
  } else {
    if (interval < DEBOUNCE_TIME) {
        queue.push(value);
    } else {
        queue.length = 0;
        queue.push(value);
    }
  }
});

А также:

// Request action from other entites
const requestAction = (x) => Rx.of(x);
// Request action from player
const requestQueueAction = () => {
    if (queue.length > 0) {
    return Rx.of(queue.shift());
  } else {
    waitingForInput = true;
    return queue$.pipe(op.take(1));
  }
}

Вот код: https://jsfiddle.net/az5pthng/7/