我们可以在nestjs中使用服务器发送的事件而不使用间隔吗？

Question

我正在使用Nestjs创建一些微服务。

例如，我有x、y和z服务，全部通过grpc互连，但我希望服务 x在特定实体更改时向 webapp 发送更新，因此我考虑了 server-sent-events [开放任何其他更好的解决方案]。

按照nestjs文档，他们有一个以n间隔运行sse路由的函数，似乎是资源耗尽的。有没有办法在有更新时实际发送事件。

假设我在同一服务中有另一个 api 调用，该调用是通过单击另一个 web 应用程序上的按钮触发的，如何触发事件仅在单击按钮时触发，而不是连续不断地发送事件。另外，如果您知道任何惯用的方法来实现这一点，那么变得hacky将不胜感激，希望它成为最后的手段。

[奖金问题]

我还考虑过使用MQTT来发送事件。但我有一种感觉，单个服务不可能拥有MQTT 和 gRPC。我对使用 MQTT 持怀疑态度，因为它的延迟以及它将如何影响内部消息传递。如果我可以限制外部客户端，那就太好了（即，使用 gRPC 进行内部连接的 x 服务和用于 web 应用程序的 MQTT 只需要 mqtt 公开一条路由）。（PS ，我是微服务新手，所以请全面介绍您的解决方案：p）

预先感谢您阅读到最后！

Answer 1

你可以。重要的是，NestJSSSE是用 Observables 实现的，所以只要你有一个可以添加的 observable，你就可以用它来发回 SSE 事件。最简单的处理方法是使用Subjects。我曾经在某个地方有过这样的例子，但一般来说，它看起来像这样

@Controller()
export class SseController {
  constructor(private readonly sseService: SseService) {}

  @SSE()
  doTheSse() {
    return this.sseService.sendEvents();
  }
}

@Injectable()
export class SseService {
  private events = new Subject();

  addEvent(event) {
    this.events.next(event);
  }

  sendEvents() {
    return this.events.asObservable();
  }
}

@Injectable()
export class ButtonTriggeredService {
  constructor(private readonly sseService: SseService) {}

  buttonClickedOrSomething() {
    this.sseService.addEvent(buttonClickedEvent);
  }
}

请原谅上面的伪代码性质，但总的来说，它确实展示了如何使用主题为 SSE 事件创建可观察值。只要@SSE()端点返回具有正确形状的可观察值，您就很成功。

Answer 2

使用NestJS的SSE有更好的处理事件的方法：

请参阅此存储库及其代码示例：

https://github.com/ningacoding/nest-sse-bug/tree/main/src

基本上你有服务的地方：

import {Injectable} from '@nestjs/common';
import {fromEvent} from "rxjs";
import {EventEmitter} from "events";

@Injectable()
export class EventsService {

    private readonly emitter = new EventEmitter();

    subscribe(channel: string) {
        return fromEvent(this.emitter, channel);
    }

    emit(channel: string, data?: object) {
        this.emitter.emit(channel, {data});
    }

}

显然，通道可以是任何字符串，建议使用路径样式。

例如：“events/for/<user_id>”，订阅该频道的用户将仅收到该频道的事件，并且仅在被触发时收到；）-

与@UseGuards等完全兼容:)

附加说明：由于已知错误，请勿在 EventsService 中注入任何服务。