How do I force an observeable to complete?

Kind of a niche question, but I know what the issue is so hopefully someone here can help me out. This is an Observable/RXFire issue, not an xstate issue.

I have this machine that invokes an observable:

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")
    }
  }
);

The way it's supposed to work is that the machine invokes the observable on load, and then once the obs is done emitting its values and calls complete(), invoke.onDone is called and the machine transitions to the 'loaded' state.

When I use a normal observable that i created with a complete() call, or when i add take(#) to the end of my .pipe(), the transition works.

But for some reason the observable that comes from collectionData() from RXFire doesn't send out a 'complete' signal... and the machine just sits there.

I've tried adding a empty() to the end and concat()-ing the observables to add a complete signal to the end of the pipe... but then I found out that empty() is deprecated and it didn't seem to work anyway.

Been banging my head against the wall for awhile. any help is appreciated.

---

Edit:

Solution:

I misunderstood the purpose of collectionData(). It is a listener, so it's not supposed to complete. I was putting a square peg in round hole. The solution is to refactor the xstate machine so I don't need to call onDone at all.

Thank you for the answers nonetheless.

---

EDIT2: GOT IT TO WORK.

take(1) can be called BEFORE concatAll(). I thought if you called it first it would end the stream, but it doesn't. The rest of the operators in the pipe still apply. So i take(1) to get the single array, use concatAll() to flatten the array into a stream of individual objects, then map that data to a new object which triggers the STORE action. the store action then sets the data to the context of the machine.

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
                         }
                        ]
                     }
            })
        }
    }
)

Thanks for everyone's help!

Observables can return multiple values over time, so it is up to collectionData() to decide when to finish (i.e. causing complete() to be called).

However, if you only want to take 1 value from the observable, you can try:

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

This will cause the observable to complete once you take 1 value from collectionData().

Note: This may not be the best solution as it depends on how the observable streams you are using works. I am just highlighting that you can use take(1) to just take 1 value and complete the source observable.