Description

This PR adds support for returning Promise<Observable<MessageEvent>> from SSE handlers, enabling async validation and custom HTTP status codes before establishing the SSE connection.

Motivation

Fixes #12260

Currently, SSE handlers can only return Observable<MessageEvent>, which means HTTP headers are sent immediately. This prevents implementing common patterns like:

• Returning 404 when a resource doesn't exist
• Returning 401 for authentication failures
• Setting custom status codes based on async validation
• Setting headers based on async data

Changes

Core Implementation

1. RouterResponseController.sse() - Now accepts Promise<Observable> in addition to Observable
2. SseStream.pipe() - Accepts optional statusCode parameter
3. RouterExecutionContext - Updated to await async SSE handlers

Key Features

• ✅ Full backward compatibility (existing code works without changes)
• ✅ Automatic detection of custom status codes via response.status()
• ✅ Works with both Express and Fastify
• ✅ Zero breaking changes

Example Usage

Resource validation with 404 response

@Sse(':gameId/events')
async gameEvents(
  @Param('gameId') gameId: string,
  @Res() response: Response,
): Promise<Observable<MessageEvent>> {
  // Async validation
  const game = await this.gameService.findById(gameId);
  
  if (!game) {
    response.status(404);
    throw new NotFoundException('Game not found');
  }
  
  return this.gameService.getGameEvents(gameId);
}

Authentication with 401 response

@Sse()
async streamNotifications(
  @Headers('authorization') authHeader: string,
  @Res() response: Response,
): Promise<Observable<MessageEvent>> {
  const user = await this.authService.validateToken(authHeader);
  
  if (!user) {
    response.status(401);
    throw new UnauthorizedException();
  }
  
  return this.notificationService.getUserNotifications(user.id);
}

Testing

Comprehensive tests added:

• ✅ Promise support
• ✅ Custom status code extraction
• ✅ Custom status code in SseStream.pipe()
• ✅ Backward compatibility
• ✅ All existing tests pass

Technical Details

• Uses Promise.resolve() for efficient handling of both sync and async cases
• Status code extracted from response object before piping
• Minimal performance overhead for existing handlers
• Type-safe implementation

Checklist

• Code follows project style guidelines
• Tests added and passing
• Non-breaking change
• Documentation in PR description