Jak rozwiązać problem z obserwowalności wewnątrz innego obserwowalnego? - rxjs

Question

Mam Observable, w którym konsumuję kolejne obserwowalne, ale drugie Observable nie mogę rozwiązać. Oto kod:

return Observable.fromPromise(axios(config)) 
     .map(res => { 
      return { 
       accessToken: res.data.access_token, 
       refreshToken: res.data.refresh_token     
      } 
     }) 
     .map(res => { 
      return { 
       me: getMe(res.accessToken), 
       accessToken: res.accessToken, 
       refreshToken: res.refreshToken     
      } 
     }) 

function getMe(accessToken) { 
    return Observable.fromPromise(axios.get({ 
     url: 'https://api.spotify.com/v1/me', 
    })); 
} 

Funkcja getMe zwraca Observable, ale nigdy nie został rozwiązany. Próbowałem dodać flatMap i concat, ale nadal nie został rozwiązany. Jak mogę rozwiązać problem z getMe?

Answer 1

Czy spróbować następujących (również niesprawdzone):

function getMe(accessToken) { 
    return Rx.Observable.fromPromise(axios.get({ 
    url: 'https://api.spotify.com/v1/me', 
    })); 
}  

Rx.Observable.fromPromise(axios(config)) 
    .map((res) => { 
     return { 
      accessToken: res.data.access_token, 
      refreshToken: res.data.refresh_token     
     } 
    }) 
    .flatMap((res) => { 
     return getMe(res.accessToken).map((res2) => { 
      res.me = res2; 
      return res; 
     } 
    }) 
    .subscribe((data) => console.log(data)); 

Jak wspomniano w powyższym poście, flatMap zwraca obserwowalne. map jest następnie używany do łączenia res z wynikiem res2 zwrócony z drugiej obietnicy.

Należy również pamiętać, że fromPromise można zaobserwować na zimno. Oznacza to, że musisz mieć abonament na inicjowanie rzeczy. W twoim przypadku, jak sądzę masz już coś takiego:

someFunction =() => { 
    return Rx.Observable.fromPromise(axios(config)) 
    ... 
    ... 
} 

someFunction.subscribe((data) => console.log(data)); 

Answer 2

Przykładowy kod do znalezienia poniżej (NIEBIESKI !!). Niektóre wyjaśnienia:

• obserwowalny zwrócony przez getMe nie jest spłaszczony („postanowienie” należą do świata obietnic), ponieważ operator map nie spłaszczyć obserwable. flatMap zrobić, ale musisz go użyć w postaci source.flatMap(function(x){return observable}), a tutaj to, co zwrócisz, to POJO, a nie Rx.Observable.
• Aby spłaszczyć getMe używamy flatMap.
• Aby dodać z powrotem brakujące pola (accessToken i refreshToken) używamy withLatestFrom na obserwowalne który emitowany obiekt res (res$).

• Używamy share, ponieważ subskrybujemy dwa razy do res$ i chcemy, aby wszyscy subskrybenci zobaczyli te same wartości.

  var res$ = Observable 
      .fromPromise(axios(config)) 
      .map(function (res) { 
       return { 
       accessToken : res.data.access_token, 
       refreshToken : res.data.refresh_token 
       } 
      }) 
      .share(); 
  var getMe$ = res$.flatMap(function (res) {return getMe(res.accessToken)}); 
  var finalRes$ = getMe$.withLatestFrom(res$, function (getMe, res) { 
  return { 
      me   : getMe, 
      accessToken : res.accessToken, 
      refreshToken : res.refreshToken 
  } 
  }); 
  
  function getMe (accessToken) { 
      return Observable.fromPromise(axios.get({url : 'https://api.spotify.com/v1/me'})); 
  } 
  

Answer 3

Jako @ user3743222 wskazał, Observable nie resolve w tym sensie, że Promise robi. Jeśli chcesz uzyskać wartość z metody getMe, musisz zasubskrybować zwracaną przez siebie wartość Observable.

return Observable.fromPromise(axios(config)) 
     .map(res => { 
      return { 
       accessToken: res.data.access_token, 
       refreshToken: res.data.refresh_token     
      } 
     }) 
     .flatMap(function(tokens) { 

      //FlatMap can implicitly accept a Promise return, so I showed that here 
      //for brevity 
      return axios.get({url : 'https://api.spotify.com/v1/me'}); 
     }, 

     //The second method gives you both the item passed into the first function 
     //paired with every item emitted from the returned Observable` 
     //i.e. axios.get(...) 
     function(tokens, response) { 
      return { 
      accessToken: tokens.accessToken, 
      refreshToken: tokens.accessToken, 
      //Here response is a value not an Observable 
      me: response 
      }; 
     });