NestJS Socket Gateway集成Mongoose Watch：ChangeStream异常关闭无变更通知

在NestJS中创建了NotificationGateway并注入notificationRepository，afterInit方法已正常执行，但MongoDB的changeStream存在两个问题：

• close事件无错误触发
• 新增Notification文档时，change事件未被触发

注：通过API执行文档增删改操作时数据库连接正常。

相关代码 #

notification.gateway.ts #

import {
  WebSocketGateway,
  WebSocketServer,
  OnGatewayInit,
  OnGatewayConnection,
  OnGatewayDisconnect,
} from '@nestjs/websockets';
import { Server } from 'socket.io';
import { InjectModel } from '@nestjs/mongoose';
import { Model } from 'mongoose';
import { Notification } from 'src/entities/notification.entity';

@WebSocketGateway({ cors: { origin: '*' } })
export class NotificationGateway
  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
{
  constructor(
    @InjectModel(Notification.name)
    private notificationsRepository: Model<Notification>,
  ) {}

  @WebSocketServer() server: Server;

  async handleNewDocument(notification: Notification) {
    console.log('Sent notificatoin', notification);
    this.server.emit('newNotification', notification);
  }

  async afterInit(server: Server) {
    console.log('WebSocket gateway initialized');

    const changeStream = this.notificationsRepository.watch();

    changeStream.on('close', (change) => {
      console.log('close');
    });

    changeStream.on('change', (change) => {
      console.log('change');
    });
  }

  async handleConnection(client: any, ...args: any[]) {
    console.log('Client connected:', client.id);
  }

  async handleDisconnect(client: any) {
    console.log('Client disconnected:', client.id);
  }
}

notification.module.ts #

import { Module } from '@nestjs/common';
import { MongooseModule } from '@nestjs/mongoose';
import {
  Notification,
  NotificationSchema,
} from 'src/entities/notification.entity';
import { NotificationGateway } from './notification.gateway';

@Module({
  imports: [
    MongooseModule.forFeature([
      { name: Notification.name, schema: NotificationSchema },
    ]),
  ],
  providers: [NotificationGateway],
  exports: [NotificationGateway],
})
export class NotificationModule {}

notification.entity.ts #

import { Prop, Schema, SchemaFactory } from '@nestjs/mongoose';
import { Transform } from 'class-transformer';
import { Types } from 'mongoose';
import { NotificationObjectType } from 'src/enums';

@Schema()
export class Notification {
  @Transform(({ value }) => value.toString())
  _id: string;

  @Prop({ required: true })
  userId: string;

  @Prop({
    required: true,
    type: String,
  })
  information: string;

  @Prop({
    required: true,
    type: Types.ObjectId,
  })
  typeId: Types.ObjectId;

  @Prop({ type: String, enum: NotificationObjectType })
  type: string;
}

export const NotificationSchema = SchemaFactory.createForClass(Notification);

NotificationSchema.set('toObject', { virtuals: true });

app.module.ts #

import { Module, MiddlewareConsumer, NestModule } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';

import { AuthModule } from './auth/auth.module';
import { UsersModule } from './user/user.module';

import { JobModule } from './job/job.module';
import { AccessControlAllowOriginMiddleware } from './middlewares/accessControlAllowOrigin.middleware';
import { MongooseModule } from '@nestjs/mongoose';
import { CandidateModule } from './candidate/candidate.module';
import { ServeStaticModule } from '@nestjs/serve-static';
import { join } from 'path';
import { NotificationModule } from './gateways/notification.module';
import { Contact, ContactSchema } from './entities/contact.entity';

@Module({
  imports: [
    AuthModule,
    UsersModule,
    JobModule,
    CandidateModule,
    NotificationModule,
    MongooseModule.forRoot(process.env.DB_URL, {
      useNewUrlParser: true,
      useUnifiedTopology: true,
    }),
    MongooseModule.forFeature([{ name: Contact.name, schema: ContactSchema }]),
    ServeStaticModule.forRoot({
      rootPath: join(__dirname, '..', 'uploads'),
    }),
  ],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule implements NestModule {
  configure(consumer: MiddlewareConsumer) {
    consumer.apply(AccessControlAllowOriginMiddleware).forRoutes('*');
  }
}

依赖配置 #

package.json dependencies #

"dependencies": {
    "@nestjs/apollo": "^10.2.0",
    "@nestjs/common": "^9.3.9",
    "@nestjs/config": "^2.3.1",
    "@nestjs/core": "^9.3.9",
    "@nestjs/graphql": "^10.2.0",
    "@nestjs/jwt": "^10.0.2",
    "@nestjs/mapped-types": "^1.2.2",
    "@nestjs/mongoose": "^9.2.1",
    "@nestjs/passport": "^9.0.3",
    "@nestjs/platform-express": "^9.3.9",
    "@nestjs/platform-socket.io": "^9.3.9",
    "@nestjs/serve-static": "^3.0.1",
    "@nestjs/websockets": "^9.3.9",
    "apollo-server-express": "^3.7.0",
    "bcrypt": "^5.1.0",
    "class-transformer": "^0.5.1",
    "class-validator": "^0.14.0",
    "cookie-parser": "^1.4.6",
    "dotenv": "^16.0.3",
    "express-session": "^1.17.3",
    "graphql": "^16.6.0",
    "mongodb": "^5.1.0",
    "mongoose": "^6.10.0",
    "passport": "^0.6.0",
    "passport-jwt": "^4.0.1",
    "passport-local": "^1.0.0",
    "pg": "^8.9.0",
    "reflect-metadata": "^0.1.13",
    "rimraf": "^4.1.2",
    "rxjs": "^7.8.0",
    "sharp": "^0.31.3",
    "ts-node-dev": "^2.0.0"
  },

环境变量 #

DB_URL=mongodb://127.0.0.1/fast-job

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