Как заставить наблюдаемое завершиться?

Question

Вроде нишевый вопрос, но я знаю, в чем проблема, поэтому надеюсь, что кто-то здесь может мне помочь. Это проблема Observable/RXFire, а не проблема xstate.

У меня есть эта машина, которая вызывает наблюдаемое:

export const tribeMachine = Machine(
  {
    id: "council",
    initial: "init",
    context: {},
    states: {
      init: {
        invoke: {
          id: "gettribes",
          src: () =>
            collectionData(database.collection("tribes")).pipe(
              concatAll(),
              map(x => ({ type: "STORE", x }))
            ),
          onDone: "loaded"
        },
        on: {
          STORE: {
            actions: "storetribes"
          },
          CANCEL: "loaded"
        }
      },
      loaded: {
        entry: () => console.log("loaded")
      },
      error: {
        entry: () => console.log("error")
      }
    }
  },
  {
    actions: {
      storetribes: (context, event) => console.log("hello")
    }
  }
);

Предполагается, что он будет работать так, что машина вызывает наблюдаемое при загрузке, а затем, когда obs завершает выдачу своих значений и вызывает complete(), вызывается invoke.onDone и машина переходит в состояние "загружено".

Когда я использую обычное наблюдаемое, которое я создал с помощью вызова complete(), или когда я добавляю take(#) в конец моего.pipe (), переход работает.

Но по какой-то причине наблюдаемое, которое поступает из collectionData() из RXFire, не отправляет сигнал "завершено"... и машина просто сидит там.

Я пробовал добавить empty() в конец и concat()- добавляя наблюдаемые, чтобы добавить полный сигнал в конец канала... но потом я обнаружил, что empty() устарел, а он не похоже, все равно работает.

Некоторое время бился головой о стену. любая помощь приветствуется.

---

Редактировать:

Решение:

Я неправильно понял цель collectionData(). Это слушатель, поэтому он не должен завершаться. Я вставлял квадратный колышек в круглое отверстие. Решение состоит в том, чтобы реорганизовать машину xstate, чтобы мне вообще не нужно было вызывать onDone.

Тем не менее, спасибо за ответы.

---

EDIT2: ДОЛЖЕН РАБОТАТЬ.

take(1) может быть вызван ДО concatAll(). Я думал, если вы сначала вызовете его, поток закончится, но это не так. Остальные операторы в трубе по-прежнему применяются. Итак, я беру (1), чтобы получить единственный массив, использую concatAll(), чтобы сгладить массив в поток отдельных объектов, затем сопоставляю эти данные с новым объектом, который запускает действие STORE. затем действие store устанавливает данные в контекст машины.

export const tribeMachine = Machine({
    id: 'council',
    initial: 'init',
    context: {
        tribes: {},
        markers: []
    },
    states: {
        init: {
            invoke: {
                id: 'gettribes',
                src: () => collectionData(database.collection('tribes')).pipe(
                    take(1),
                    concatAll(),
                    map(value => ({ type: 'TRIBESTORE', value })),
                ),
                onDone: 'loaded'
            },
            on: {
                TRIBESTORE: {
                    actions: ['storetribes', 'logtribes']
                },
                CANCEL: 'loaded'
            }
        },
        loaded: {
        },
        error: {
        }
    }
},
    {
        actions: {
            storetribes: assign((context, event) => {
                return {
                    tribes: {
                        ...context.tribes,
                        [event.value.id]: event.value
                     },
                     markers: [
                         ...context.markers,
                         {
                             lat: event.value.lat,
                             lng: event.value.lng,
                             title: event.value.tribeName
                         }
                        ]
                     }
            })
        }
    }
)

Спасибо всем за помощь!

Answer 1

Наблюдаемые могут возвращать несколько значений с течением времени, так что до collectionData() чтобы решить, когда закончить (т.е. вызвать complete()).

Однако, если вы хотите взять только 1 значение из наблюдаемого, вы можете попробовать:

  collectionData(database.collection("tribes")).pipe(
              take(1),
              concatAll(),
              map(x => ({ type: "STORE", x }))
            ),

Это приведет к завершению наблюдаемого, как только вы возьмете 1 значение из collectionData().

Примечание: это может быть не лучшим решением, поскольку это зависит от того, как работают наблюдаемые потоки, которые вы используете. Я просто подчеркиваю, что вы можете использоватьtake(1) просто взять 1 значение и завершить наблюдаемый источник.