我已经使用以下方法在 Nestjs 中实现了 WebSockets,其中一个要求是,一旦我收到来自 websocket 服务器的任何消息,我就需要更新数据库。为了做到这一点,我已经完成了下面的代码,但得到了这样的错误:
但是,如果我在任何其他控制器方法中调用相同的方法,它就可以正常工作(检查 /test1)。只有当我从 websocket 侦听器调用它时,我才会收到此错误。即使我从同一个控制器调用简单方法来打印一些日志,我也会收到相同的错误TypeError: this.processData is not a function。诡异的!
有人可以告诉我我做错了什么吗?
(node:11421) UnhandledPromiseRejectionWarning: TypeError: Cannot read property 'updateResponseData' of undefined
at WebSocket.incoming (/opt/dist/controller/sample.controller.js:62:44)
at WebSocket.emit (events.js:210:5)
at Receiver.receiverOnMessage (/opt/node_modules/ws/lib/websocket.js:800:20)
at Receiver.emit (events.js:210:5)
at Receiver.dataMessage (/opt/node_modules/ws/lib/receiver.js:423:14)
at Receiver.getData (/opt/node_modules/ws/lib/receiver.js:353:17)
at Receiver.startLoop (/opt/node_modules/ws/lib/receiver.js:139:22)
at Receiver._write (/opt/node_modules/ws/lib/receiver.js:74:10)
at doWrite (_stream_writable.js:431:12)
at writeOrBuffer (_stream_writable.js:415:5)
at Receiver.Writable.write (_stream_writable.js:305:11)
at Socket.socketOnData (/opt/node_modules/ws/lib/websocket.js:875:35)
at Socket.emit (events.js:210:5)
at addChunk (_stream_readable.js:308:12)
at readableAddChunk (_stream_readable.js:289:11)
at Socket.Readable.push (_stream_readable.js:223:10)
(node:11421) UnhandledPromiseRejectionWarning: Unhandled promise rejection. This error originated either by throwing inside of an async function without a catch block, or by rejecting a promise which was not handled with .catch(). (rejection id: 1)
(node:11421) [DEP0018] DeprecationWarning: Unhandled promise rejections are deprecated. In the future, promise rejections that are not handled will terminate the Node.js process with a non-zero exit code.
控制器代码:
import { Controller, Post, Get, Inject, Req, UsePipes, Body, Header,Param } from '@nestjs/common';
import { Request } from 'express';
import * as WebSocket from 'ws';
import { DatabaseService } from '../service/datafetch.service';
@Controller('sample')
export class SampleController {
@Inject(dbService)
private dbService: DatabaseService;
static ws: any;
static wsnew: any;
public receiveDistData = '';
// Connect WS & Listen to messages
constructor() {
this.initWs();
SampleController.ws.on('message', async function incoming(data) {
console.log('message recieved 8081');
var dataJson = JSON.parse(data);
await this.dbService.updateResponseData(dataJson);
return data;
});
}
@Get('/test/data/:id')
async testData(@Param('id') id: string) {
return await this.callWs(id);
}
@Get('/test1/data/:id')
async testData1(@Param('id') id: string) {
const data = {id: id, data:{}};
return await this.dbService.updateResponseData(data);
}
async initWs() {
SampleController.ws = new WebSocket('ws://127.0.0.1:8081');
}
async processData() {
console.log('Printing a log...');
}
// Data Distribution
async callWs(id) {
// If Socket is not opened, try to re-open
if(SampleController.ws.readyState != 1) {
console.log('Server is dead....');
this.initWs();
}
const Data = await this.dbService.findAll(id);
await SampleController.ws.send(JSON.stringify({
event: 'channel1',
data: Data,
}));
}
}
存储库代码:
import { InjectRepository } from '@nestjs/typeorm';
import { Injectable } from '@nestjs/common';
import { Repository } from 'typeorm';
import { SampleRepository } from '../repository/sample.repository';
@Injectable()
export class SampleService {
constructor(
@InjectRepository(SampleRepository)
private readonly sampleRepository: SampleRepository
) {}
async updateResponseData(data): Promise<any> {
return await this.sampleRepository.updateData(data);
}
}