在rxjs中连接两个observable时无法接收所有数据

Question

我从数据库中创建了两个observable：

const roomData$ = estab
      .query()
      .where('estab.type', 'like', 'rooms')
      .then((rooms) => {
        rooms.forEach((room) => {
          observer.next(room.data);
        });
      });
  }).map(data => ({
    roomsData: [
      ...data.rooms,
    ],
  }));
const hotelData$ = Observable.create((observer) => {
        estab
          .query()
          .where('estab.type', 'like', 'hotels')
          .then((hotels) => {
            hotels.forEach((hotel) => {
              observer.next(hotel.data);
            });
          });
      }).map(data => ({
        hotelsData: [
          ...data.hotels,
        ],
      }));  
const hotelContentData$ = Observable.concat(roomData$, hotelData$);
hotelContentData$.subscribe((data) =>  {
    fs.appendFile('data.json', (err) => { 
         if (err) throw err;
         console.log('file has been appended!);
   }      
})

在data.json文件中，我发现只有从roomData$可观察到的数据。我改变了Observable.concat(hotelData$, roomData$)的顺序，现在我得到了酒店可观察的数据。为什么只执行一个Observable？如何执行这两个并从hotelData$和roomData$获取数据？

Answer 1

Observable.concat将从流中获取排放，直到该流完成，然后移至下一个流。由于您正在创建自己的可观察序列，请务必完成每个序列。

const roomData$ = Observable.create( observer => {
  estab
    .query()
    .where('estab.type', 'like', 'rooms')
    .then((rooms) => {
      rooms.forEach((room) => {
        observer.next(room.data);
      });
      observer.complete(); // adding this will complete your stream.
    });
}).map(data => ({
  roomsData: [
    ...data.rooms,
  ],
}));

const hotelData$ = Observable.create((observer) => {
  estab
    .query()
    .where('estab.type', 'like', 'hotels')
    .then((hotels) => {
      hotels.forEach((hotel) => {
        observer.next(hotel.data);
      });
      observer.complete(); // adding this will complete your stream.
    });
}).map(data => ({
  hotelsData: [
    ...data.hotels,
  ],
}));

const hotelContentData$ = Observable.concat(roomData$, hotelData$);
hotelContentData$.subscribe((data) =>  {
  fs.appendFile('data.json', (err) => { 
    if (err) throw err;
    console.log('file has been appended!);
  });      
});