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

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.

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

---

РЕДАКТИРОВАТЬ 2: ПОЛУЧИЛ ЭТО РАБОТАТЬ.

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() можно решить, когда завершить sh (то есть вызывать метод complete ()).

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

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

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

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