amirho/nest
1
0
Fork 0
mirror of https://github.com/nestjs/nest.git synced 2026-02-21 23:11:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(): resolve merge conflicts

Browse Source
This commit is contained in:
Kamil Myśliwiec
2020-02-20 13:08:39 +01:00
parent 75659ab076 3cf1924cde
commit 01f7bb6949
854 changed files with 335677 additions and 14243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
packages/microservices/server/server-mqtt.ts
View File

@@ -7,6 +7,7 @@ import {
MQTT_DEFAULT_URL,
NO_MESSAGE_HANDLER,
} from '../constants';
import { MqttContext } from '../ctx-host/mqtt.context';
import { MqttClient } from '../external/mqtt-client.interface';
import {
CustomTransportStrategy,
@@ -67,19 +68,24 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
}
public getMessageHandler(pub: MqttClient): Function {
return async (channel: string, buffer: Buffer) =>
this.handleMessage(channel, buffer, pub);
return async (
channel: string,
buffer: Buffer,
originalPacket?: Record<string, any>,
) => this.handleMessage(channel, buffer, pub, originalPacket);
}
public async handleMessage(
channel: string,
buffer: Buffer,
pub: MqttClient,
originalPacket?: Record<string, any>,
): Promise<any> {
const rawPacket = this.parseMessage(buffer.toString());
const packet = this.deserializer.deserialize(rawPacket, { channel });
const mqttContext = new MqttContext([channel, originalPacket]);
if (isUndefined((packet as IncomingRequest).id)) {
return this.handleEvent(channel, packet);
return this.handleEvent(channel, packet, mqttContext);
}
const publish = this.getPublisher(
pub,
@@ -98,7 +104,7 @@ export class ServerMqtt extends Server implements CustomTransportStrategy {
return publish(noHandlerPacket);
}
const response$ = this.transformToObservable(
await handler(packet.data),
await handler(packet.data, mqttContext),
) as Observable<any>;
response$ && this.send(response$, publish);
}