Custom transportyorlar

Nest qutidan turli xil transporterslarni taqdim etadi, shuningdek, yangi custom transport strategiyalarini yaratish imkonini beruvchi API ham bor. Transportyorlar komponentlarni tarmoq orqali ulanadigan, pluggable aloqa qatlami va juda sodda ilova darajasidagi xabar protokoli orqali bog'lash imkonini beradi (to'liq articleni o'qing).

Custom transporter bilan siz istalgan messaging tizimi/protokolini (jumladan Google Cloud Pub/Sub, Amazon Kinesis va boshqalar) integratsiya qilishingiz yoki mavjudini kengaytirib, qo'shimcha imkoniyatlar qo'shishingiz mumkin (masalan, MQTT uchun QoS).

Strategiya yaratish

Avval custom transportyerimizni ifodalovchi klassni aniqlaymiz.

1import { CustomTransportStrategy, Server } from '@nestjs/microservices';
2
3class GoogleCloudPubSubServer
4  extends Server
5  implements CustomTransportStrategy
6{
7  /**
8   * Triggered when you run "app.listen()".
9   */
10  listen(callback: () => void) {
11    callback();
12  }
13
14  /**
15   * Triggered on application shutdown.
16   */
17  close() {}
18
19  /**
20   * You can ignore this method if you don't want transporter users
21   * to be able to register event listeners. Most custom implementations
22   * will not need this.
23   */
24  on(event: string, callback: Function) {
25    throw new Error('Method not implemented.');
26  }
27
28  /**
29   * You can ignore this method if you don't want transporter users
30   * to be able to retrieve the underlying native server. Most custom implementations
31   * will not need this.
32   */
33  unwrap<T = never>(): T {
34    throw new Error('Method not implemented.');
35  }
36}

Yuqoridagi misolda GoogleCloudPubSubServer klassini e'lon qildik va CustomTransportStrategy interfeysi majbur qiladigan listen() va close() metodlarini taqdim etdik. Shuningdek, klassimiz @nestjs/microservices paketidan import qilingan Server klassini kengaytiradi; u Nest runtime tomonidan message handlerlarni ro'yxatdan o'tkazishda foydalaniladigan foydali metodlarni taqdim etadi. Muqobil ravishda, mavjud transport strategiyasi imkoniyatlarini kengaytirmoqchi bo'lsangiz, mos server klassini, masalan ServerRedisni kengaytirishingiz mumkin. An'anaviy tarzda, biz klass nomiga "Server" suffiksini qo'shdik, chunki u xabarlar/eventlarga subscribe qilish (va kerak bo'lsa ularga javob berish) uchun javobgar bo'ladi.

Shu bilan, endi built-in transportyor o'rniga custom strategiyamizdan quyidagicha foydalanishimiz mumkin:

1const app = await NestFactory.createMicroservice<MicroserviceOptions>(
2  AppModule,
3  {
4    strategy: new GoogleCloudPubSubServer(),
5  },
6);

Aslida, transport va options propertylari bilan odatdagi transport options obyektini berish o'rniga, biz bitta strategy propertysini uzatamiz va uning qiymati custom transportyer klassining instansiyasi bo'ladi.

GoogleCloudPubSubServer klassimizga qaytsak, real ilovada biz listen() metodida message broker/tashqi servisga ulanishni o'rnatib, subscriberlarni ro'yxatdan o'tkazardik (so'ng close() teardown metodida obunalarni bekor qilib, ulanishni yopardik), lekin bu Nest microservice'lar bir-biri bilan qanday muloqot qilishini yaxshi tushunishni talab qiladi, shuning uchun ushbu article seriesni o'qishni tavsiya qilamiz. Bu bobda esa Server klassi taqdim etadigan imkoniyatlar va ularni custom strategiyalar qurishda qanday ishlatishingiz mumkinligiga e'tibor qaratamiz.

Masalan, ilovamizning biror joyida quyidagi message handler aniqlangan bo'lsin:

1@MessagePattern('echo')
2echo(@Payload() data: object) {
3  return data;
4}

Bu message handler Nest runtime tomonidan avtomatik ro'yxatdan o'tkaziladi. Server klassi bilan siz qaysi message patternlar ro'yxatdan o'tganini ko'rishingiz va ularga biriktirilgan metodlarga kirish hamda ularni bajarishingiz mumkin. Buni sinab ko'rish uchun listen() metodida callback funksiyasi chaqirilishidan oldin oddiy console.log qo'shamiz:

1listen(callback: () => void) {
2  console.log(this.messageHandlers);
3  callback();
4}

Ilova qayta ishga tushgandan so'ng terminalda quyidagi logni ko'rasiz:

1Map { 'echo' => [AsyncFunction] { isEventHandler: false } }

Ko'rib turganingizdek, messageHandlers propertysi barcha message (va event) handlerlarining Map kolleksiyasidir, unda patternlar key sifatida ishlatiladi. Endi, masalan, "echo" kabi key yordamida message handlerga referensni olishingiz mumkin:

1async listen(callback: () => void) {
2  const echoHandler = this.messageHandlers.get('echo');
3  console.log(await echoHandler('Hello world!'));
4  callback();
5}

echoHandlerni ixtiyoriy string argument bilan chaqirgach (bu yerda "Hello world!"), uni konsolda ko'rishimiz kerak:

1Hello world!

Bu handler metodimiz to'g'ri bajarilganini anglatadi.

Interceptors bilan CustomTransportStrategy ishlatilganda, handlerlar RxJS streamlariga o'raladi. Bu, streamning ichki mantiqi bajarilishi uchun ularga subscribe bo'lishingiz kerakligini anglatadi (masalan, interceptor bajarilgandan so'ng controller mantiqiga o'tish).

Bunga misol quyida ko'rsatilgan:

1async listen(callback: () => void) {
2  const echoHandler = this.messageHandlers.get('echo');
3  const streamOrResult = await echoHandler('Hello World');
4  if (isObservable(streamOrResult)) {
5    streamOrResult.subscribe();
6  }
7  callback();
8}

Client proksi

Birinchi bo'limda aytib o'tganimizdek, microservice yaratish uchun @nestjs/microservices paketidan foydalanishingiz shart emas, lekin agar shunday qilishni xohlasangiz va custom strategiyani integratsiya qilmoqchi bo'lsangiz, sizga "client" klassi ham kerak bo'ladi.

Tashqi servis bilan muloqot qilish/xabarlarni emit va publish qilish (yoki eventlar) uchun siz kutubxonaga xos SDK paketidan foydalanishingiz yoki ClientProxyni kengaytiradigan custom client klassini quyidagicha implement qilishingiz mumkin:

1import { ClientProxy, ReadPacket, WritePacket } from '@nestjs/microservices';
2
3class GoogleCloudPubSubClient extends ClientProxy {
4  async connect(): Promise<any> {}
5  async close() {}
6  async dispatchEvent(packet: ReadPacket<any>): Promise<any> {}
7  publish(
8    packet: ReadPacket<any>,
9    callback: (packet: WritePacket<any>) => void,
10  ): Function {}
11  unwrap<T = never>(): T {
12    throw new Error('Method not implemented.');
13  }
14}

Ko'rib turganingizdek, ClientProxy klassi bizdan ulanishni o'rnatish va yopish hamda xabarlarni (publish) va eventlarni (dispatchEvent) yuborish uchun bir nechta metodlarni taqdim etishni talab qiladi. Eslatma: request-response muloqotini qo'llab-quvvatlash shart bo'lmasa, publish() metodini bo'sh qoldirishingiz mumkin. Xuddi shuningdek, event-based muloqotni qo'llab-quvvatlash kerak bo'lmasa, dispatchEvent() metodini tashlab keting.

Ushbu metodlar qachon va nimalarni bajarishini ko'rish uchun quyidagicha bir nechta console.log qo'shamiz:

1class GoogleCloudPubSubClient extends ClientProxy {
2  async connect(): Promise<any> {
3    console.log('connect');
4  }
5
6  async close() {
7    console.log('close');
8  }
9
10  async dispatchEvent(packet: ReadPacket<any>): Promise<any> {
11    return console.log('event to dispatch: ', packet);
12  }
13
14  publish(
15    packet: ReadPacket<any>,
16    callback: (packet: WritePacket<any>) => void,
17  ): Function {
18    console.log('message:', packet);
19
20    // In a real-world application, the "callback" function should be executed
21    // with payload sent back from the responder. Here, we'll simply simulate (5 seconds delay)
22    // that response came through by passing the same "data" as we've originally passed in.
23    //
24    // The "isDisposed" bool on the WritePacket tells the response that no further data is
25    // expected. If not sent or is false, this will simply emit data to the Observable.
26    setTimeout(() => callback({ 
27      response: packet.data,
28      isDisposed: true,
29    }), 5000);
30
31    return () => console.log('teardown');
32  }
33
34  unwrap<T = never>(): T {
35    throw new Error('Method not implemented.');
36  }
37}

Shu bilan, GoogleCloudPubSubClient klassi instansiyasini yaratib, (oldingi boblarda ko'rganingiz kabi) send() metodini ishga tushiramiz va qaytgan observable streamga subscribe bo'lamiz.

1const googlePubSubClient = new GoogleCloudPubSubClient();
2googlePubSubClient
3  .send('pattern', 'Hello world!')
4  .subscribe((response) => console.log(response));

Endi terminalda quyidagi chiqishni ko'rasiz:

1connect
2message: { pattern: 'pattern', data: 'Hello world!' }
3Hello world! // <-- after 5 seconds

"teardown" metodimiz (publish() metodi qaytaradigan) to'g'ri bajarilayotganini tekshirish uchun streamga timeout operatorini qo'llaymiz va uni 2 soniyaga o'rnatamiz. Bu setTimeout callbackni chaqirishidan oldin xatoni chiqarishini ta'minlaydi.

1const googlePubSubClient = new GoogleCloudPubSubClient();
2googlePubSubClient
3  .send('pattern', 'Hello world!')
4  .pipe(timeout(2000))
5  .subscribe(
6    (response) => console.log(response),
7    (error) => console.error(error.message),
8  );