NestJS+RxJS递归调用YouTube API获取视频失败问题

Let's break down why your current code isn't pulling in subsequent pages of videos, then fix it with a clean, working implementation.

What's Wrong With the Original Code? #

Your core issue is a misuse of RxJS operators:

• The tap operator only handles side effects (like logging) — it doesn't modify or replace the Observable stream. Any mergeMap or empty() you return inside tap is completely ignored, so your subsequent page requests never actually run.
• You had a parameter name bug: you referenced video['nextPageToken'] instead of videoResponse['nextPageToken'] in your mergeMap.
• toPromise() is deprecated in RxJS 7+; we should use modern alternatives like firstValueFrom.

Fixed Implementation (RxJS-Based) #

Here's a corrected version that properly handles recursive pagination:

import { Injectable } from '@nestjs/common';
import { HttpService } from '@nestjs/axios';
import { firstValueFrom, empty, mergeMap } from 'rxjs';
import { map, tap } from 'rxjs/operators';

@Injectable()
export class YoutubeService {
  private videos: any[] = [];

  constructor(private readonly httpService: HttpService) {}

  private getPlaylistURL(playlistId: string, pageToken?: string): string {
    // Replace with your actual YouTube API URL construction logic
    const baseUrl = 'https://www.googleapis.com/youtube/v3/playlistItems';
    const apiKey = 'YOUR_YOUTUBE_API_KEY'; // Add your API key here
    let url = `${baseUrl}?part=snippet&maxResults=50&playlistId=${playlistId}&key=${apiKey}`;
    if (pageToken) url += `&pageToken=${pageToken}`;
    return url;
  }

  async getPlaylistVideos(playlistId: string): Promise<any[]> {
    this.videos = [];

    // Recursive function to fetch a single page and trigger next page if needed
    const fetchPage = (pageToken?: string) => {
      return this.httpService.get(this.getPlaylistURL(playlistId, pageToken)).pipe(
        map(response => response.data),
        tap(videoResponse => {
          console.log(`\tFetched page with token: ${pageToken ?? 'none'}`);
          this.videos.push(...videoResponse.items);
        }),
        mergeMap(videoResponse => {
          if (videoResponse.nextPageToken) {
            console.log(`\tFetching next page with token: ${videoResponse.nextPageToken}`);
            return fetchPage(videoResponse.nextPageToken);
          } else {
            console.log('\tNo more pages to fetch');
            return empty();
          }
        })
      );
    };

    try {
      await firstValueFrom(fetchPage());
      return this.videos;
    } catch (err) {
      console.error('Failed to fetch playlist videos:', err);
      throw err;
    }
  }
}

Key Improvements #

1. Moved pagination logic to mergeMap: This operator replaces the current Observable with a new one (the next page request) when a nextPageToken exists, enabling proper recursive fetching.
2. Used firstValueFrom: Replaces the deprecated toPromise() with RxJS's recommended approach for converting Observables to Promises.
3. Cleaner recursive structure: The fetchPage function encapsulates single-page logic, making the code easier to follow and debug.
4. Fixed parameter name bug: Properly references videoResponse.nextPageToken instead of the incorrect video variable.

Alternative: Async/Await Only (No RxJS Operators) #

If you prefer a simpler approach without RxJS operators (more aligned with NestJS's async/await style), here's an even cleaner version:

async getPlaylistVideos(playlistId: string): Promise<any[]> {
  const videos: any[] = [];

  const fetchPage = async (pageToken?: string) => {
    const response = await firstValueFrom(
      this.httpService.get(this.getPlaylistURL(playlistId, pageToken))
    );
    const videoResponse = response.data;
    
    videos.push(...videoResponse.items);
    console.log(`\tFetched ${videoResponse.items.length} videos | Total: ${videos.length}`);

    if (videoResponse.nextPageToken) {
      await fetchPage(videoResponse.nextPageToken);
    }
  };

  try {
    await fetchPage();
    return videos;
  } catch (err) {
    console.error('Failed to fetch playlist videos:', err);
    throw err;
  }
}

This version avoids RxJS operators entirely, using async/await recursion to handle pagination — it's often easier to reason about for simple recursive workflows.

内容的提问来源于stack exchange，提问作者Sankalp