NestJs @Sse -事件仅由一个客户端使用。开发者社区

相关文章推荐

不爱学习的生姜 · Electron将Vue项目封装为本地应用_ ...· 4 小时前 ·

安静的李子 · 请求确认存在劳动关系案件的管辖确定与处理· 1 周前 ·

率性的黄瓜 · 发布了！来认识赣南客家菜“十大名菜”“十大名 ...· 9 月前 ·

直爽的萝卜 · 北京新机场线基础票价不超36.5元 ...· 10 月前 ·

性感的卤蛋 · 中心黎沛文老师参加第27届华夏园丁大联欢活动 ...· 1 年前 ·

任性的跑步鞋 · 徐芑南交大成就我的深潜人生-校史网· 1 年前 ·

腾讯云

开发者社区

文档意见反馈控制台

首页 TVP

文章/答案/技术大牛

发布

  @Sse('sse')
  sse(): Observable<MessageEvent> {
    return interval(5000).pipe(
      map((_) => ({ data: { hello: `world - ${this.c++}` }} as MessageEvent)),
  }

import { Controller, Get, MessageEvent, OnModuleDestroy, OnModuleInit, Res, Sse } from '@nestjs/common';
import { readFileSync } from 'fs';
import { join } from 'path';
import { Observable, ReplaySubject } from 'rxjs';
import { map } from 'rxjs/operators';
import { Response } from 'express';
@Controller()
export class AppController implements OnModuleInit, OnModuleDestroy {
  private stream: {
    id: string;
    subject: ReplaySubject<unknown>;
    observer: Observable<unknown>;
  }[] = [];
  private timer: NodeJS.Timeout;
  private id = 0;
  public onModuleInit(): void {
    this.timer = setInterval(() => {
      this.id += 1;
      this.stream.forEach(({ subject }) => subject.next(this.id));
    }, 1000);
  public onModuleDestroy(): void {
    clearInterval(this.timer);
  @Get()
  public index(): string {
    return readFileSync(join(__dirname, 'index.html'), 'utf-8').toString();
  @Sse('sse')
  public sse(@Res() response: Response): Observable<MessageEvent> {
    const id = AppController.genStreamId();
    // Clean up the stream when the client disconnects
    response.on('close', () => this.removeStream(id));
    // Create a new stream
    const subject = new ReplaySubject();
    const observer = subject.asObservable();
    this.addStream(subject, observer, id);
    return observer.pipe(map((data) => ({
      id: `my-stream-id:${id}`,
      data: `Hello world ${data}`,
      event: 'my-event-name',
    }) as MessageEvent));
  private addStream(subject: ReplaySubject<unknown>, observer: Observable<unknown>, id: string): void {
    this.stream.push({
      subject,
      observer,
  private removeStream(id: string): void {
    this.stream = this.stream.filter(stream => stream.id !== id);
  private static genStreamId(): string {
    return Math.random().toString(36).substring(2, 15);
}

 private events: Subject<MessageEvent> = new Subject();
 constuctor(){
   timer(0, 1000).pipe(takeUntil(this.destroy)).subscribe(async (index: any)=>{
            let event: MessageEvent = { 
                id: index, 
                type: 'test', 
                retry: 30000, 
                data: {index: index}
            } as MessageEvent;
            this.events.next(event);

回答 3

Stack Overflow用户

Stack Overflow用户

Stack Overflow用户