Custom transporters (transporter tùy chỉnh)

Nest cung cấp sẵn nhiều transporter khác nhau, cùng với một API cho phép bạn tự xây dựng chiến lược transport (transport strategy) tùy chỉnh. Transporter giúp bạn kết nối các component qua network bằng một lớp giao tiếp (communications layer) có thể cắm ghép (pluggable) và một message protocol ở mức application rất đơn giản (đọc thêm bài viết đầy đủ).

Với custom transporter, bạn có thể tích hợp bất kỳ messaging system/protocol nào (bao gồm Google Cloud Pub/Sub, Amazon Kinesis, v.v…) hoặc mở rộng transporter có sẵn, thêm các tính năng riêng (ví dụ QoS cho MQTT).

Creating a strategy (tạo strategy)

Trước tiên, hãy định nghĩa một class đại diện cho custom transporter của chúng ta.

import { CustomTransportStrategy, Server } from '@nestjs/microservices';

class GoogleCloudPubSubServer
  extends Server
  implements CustomTransportStrategy
{
  /**
   * Triggered when you run "app.listen()".
   */
  listen(callback: () => void) {
    callback();
  }

  /**
   * Triggered on application shutdown.
   */
  close() {}

  /**
   * You can ignore this method if you don't want transporter users
   * to be able to register event listeners. Most custom implementations
   * will not need this.
   */
  on(event: string, callback: Function) {
    throw new Error('Method not implemented.');
  }

  /**
   * You can ignore this method if you don't want transporter users
   * to be able to retrieve the underlying native server. Most custom implementations
   * will not need this.
   */
  unwrap<T = never>(): T {
    throw new Error('Method not implemented.');
  }
}

Trong ví dụ trên, chúng ta khai báo class GoogleCloudPubSubServer và cung cấp hai method listen() và close() mà interface CustomTransportStrategy bắt buộc phải có. Class này còn extend từ class Server được import từ package @nestjs/microservices, cung cấp sẵn một số method hữu ích, ví dụ các method được Nest runtime dùng để đăng ký message handler. Ngoài ra, nếu bạn muốn mở rộng khả năng của một transport strategy có sẵn, bạn có thể extend từ class server tương ứng, ví dụ ServerRedis. Theo quy ước, chúng ta thêm hậu tố "Server" vào tên class vì nó sẽ chịu trách nhiệm subscribe message/event (và phản hồi lại nếu cần).

Sau khi có class này, chúng ta có thể dùng strategy tùy chỉnh của mình thay cho một transporter có sẵn, như sau:

const app = await NestFactory.createMicroservice<MicroserviceOptions>(
  AppModule,
  {
    strategy: new GoogleCloudPubSubServer(),
  },
);

Về cơ bản, thay vì truyền object options thông thường gồm transport và options, chúng ta truyền một property duy nhất là strategy, với giá trị là một instance của custom transporter class.

Trở lại với class GoogleCloudPubSubServer, trong một ứng dụng thực tế, chúng ta sẽ thiết lập kết nối đến message broker/external service và đăng ký subscriber/lắng nghe các channel cụ thể trong method listen() (sau đó gỡ subscription & đóng kết nối trong method teardown close()), nhưng vì việc này yêu cầu hiểu rõ cách các microservice của Nest giao tiếp với nhau, chúng tôi khuyến khích đọc series bài viết này. Ở chương này, chúng ta sẽ tập trung vào các khả năng mà class Server cung cấp và cách bạn có thể tận dụng chúng để xây dựng custom strategy.

Ví dụ, giả sử ở đâu đó trong ứng dụng có định nghĩa message handler sau:

@MessagePattern('echo')
echo(@Payload() data: object) {
  return data;
}

Message handler này sẽ được Nest runtime tự động đăng ký. Với class Server, bạn có thể xem những message pattern nào đã được đăng ký, và cũng có thể truy cập, thực thi trực tiếp các method đã được gán cho pattern đó. Để kiểm tra điều này, hãy thêm một console.log đơn giản bên trong method listen(), trước khi hàm callback được gọi:

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

Sau khi restart ứng dụng, bạn sẽ thấy log sau trong terminal:

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

Như bạn thấy, property messageHandlers là một Map chứa tất cả message (và event) handler, trong đó pattern được dùng làm key. Bây giờ, bạn có thể dùng một key (ví dụ "echo") để lấy reference tới message handler:

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

Sau khi thực thi echoHandler với một chuỗi bất kỳ làm argument (ở đây là "Hello world!"), chúng ta sẽ thấy kết quả trong console:

Hello world!

Điều này cho thấy method handler của chúng ta đã được thực thi đúng.

Khi dùng CustomTransportStrategy cùng với Interceptors, các handler sẽ được bọc trong RxJS stream. Nghĩa là bạn cần subscribe vào chúng để logic bên trong stream được thực thi (ví dụ: tiếp tục vào logic của controller sau khi interceptor đã chạy xong).

Ví dụ dưới đây minh họa điều này:

async listen(callback: () => void) {
  const echoHandler = this.messageHandlers.get('echo');
  const streamOrResult = await echoHandler('Hello World');
  if (isObservable(streamOrResult)) {
    streamOrResult.subscribe();
  }
  callback();
}

Reporting event handler errors (báo lỗi từ event handler)

Event handler không có response stream, nên một exception ném ra bên trong nó không có nơi nào để đi tiếp. Nest sẽ tự log lại lỗi đó, nhưng chỉ sau khi các exception filter của bạn đã chạy. Nếu một filter khớp với exception đó, filter đó sẽ là nơi duy nhất báo lỗi.

Một số transporter tự await event stream và chuyển lỗi đó cho client library của chúng để library đó tự báo lỗi. ServerKafka làm theo cách này: kafkajs sẽ tự log lại rejection, nên nếu Nest cũng log thì sẽ bị trùng lặp. Trong trường hợp đó, đặt propagatesEventHandlerErrors thành true:

class GoogleCloudPubSubServer
  extends Server
  implements CustomTransportStrategy
{
  /**
   * The client library reports an event handler failure, so Nest must not
   * log it a second time.
   */
  public override readonly propagatesEventHandlerErrors = true;
}

Giá trị mặc định là false, phù hợp với transporter dùng handleEvent được kế thừa sẵn. Trong trường hợp đó không có gì subscribe vào event stream, nên Nest là nơi duy nhất có thể báo lỗi.

Client proxy

Như đã nói ở phần đầu, bạn không nhất thiết phải dùng package @nestjs/microservices để tạo microservice, nhưng nếu bạn quyết định dùng nó và cần tích hợp một custom strategy, bạn cũng cần cung cấp thêm một “client” class.

Để giao tiếp với external service/emit & publish message (hoặc event), bạn có thể dùng SDK package riêng của thư viện đó, hoặc tự implement một custom client class bằng cách extend ClientProxy, như sau:

import { ClientProxy, ReadPacket, WritePacket } from '@nestjs/microservices';

class GoogleCloudPubSubClient extends ClientProxy {
  async connect(): Promise<any> {}
  async close() {}
  async dispatchEvent(packet: ReadPacket<any>): Promise<any> {}
  publish(
    packet: ReadPacket<any>,
    callback: (packet: WritePacket<any>) => void,
  ): Function {}
  unwrap<T = never>(): T {
    throw new Error('Method not implemented.');
  }
}

Như bạn thấy, class ClientProxy yêu cầu chúng ta cung cấp một số method để thiết lập & đóng kết nối, cùng với việc publish message (publish) và event (dispatchEvent). Lưu ý, nếu bạn không cần hỗ trợ kiểu giao tiếp request-response, bạn có thể để method publish() trống. Tương tự, nếu không cần hỗ trợ giao tiếp dựa trên event, có thể bỏ qua method dispatchEvent().

Để quan sát các method này được thực thi khi nào và như thế nào, hãy thêm vài console.log, như sau:

class GoogleCloudPubSubClient extends ClientProxy {
  async connect(): Promise<any> {
    console.log('connect');
  }

  async close() {
    console.log('close');
  }

  async dispatchEvent(packet: ReadPacket<any>): Promise<any> {
    return console.log('event to dispatch: ', packet);
  }

  publish(
    packet: ReadPacket<any>,
    callback: (packet: WritePacket<any>) => void,
  ): Function {
    console.log('message:', packet);

    // In a real-world application, the "callback" function should be executed
    // with payload sent back from the responder. Here, we'll simply simulate (5 seconds delay)
    // that response came through by passing the same "data" as we've originally passed in.
    //
    // The "isDisposed" bool on the WritePacket tells the response that no further data is
    // expected. If not sent or is false, this will simply emit data to the Observable.
    setTimeout(() => callback({ 
      response: packet.data,
      isDisposed: true,
    }), 5000);

    return () => console.log('teardown');
  }

  unwrap<T = never>(): T {
    throw new Error('Method not implemented.');
  }
}

Với đoạn code trên, hãy tạo một instance của class GoogleCloudPubSubClient và chạy method send() (mà có thể bạn đã thấy ở các chương trước), subscribe vào observable stream trả về.

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

Bây giờ, bạn sẽ thấy output sau trong terminal:

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

Để kiểm tra method “teardown” (được method publish() của chúng ta trả về) có được thực thi đúng không, hãy áp dụng operator timeout vào stream, đặt là 2 giây để chắc chắn nó throw trước khi setTimeout gọi hàm callback.

const googlePubSubClient = new GoogleCloudPubSubClient();
googlePubSubClient
  .send('pattern', 'Hello world!')
  .pipe(timeout(2000))
  .subscribe(
    (response) => console.log(response),
    (error) => console.error(error.message),
  );

Với operator timeout được áp dụng, output trong terminal sẽ như sau:

connect
message: { pattern: 'pattern', data: 'Hello world!' }
teardown // <-- teardown
Timeout has occurred

Để dispatch một event (thay vì gửi message), dùng method emit():

googlePubSubClient.emit('event', 'Hello world!');

Và đây là những gì bạn sẽ thấy trong console:

connect
event to dispatch:  { pattern: 'event', data: 'Hello world!' }

Message serialization (tuần tự hóa message)

Nếu bạn cần thêm logic tùy chỉnh xung quanh việc serialization của response ở phía client, bạn có thể dùng một custom class extend từ class ClientProxy hoặc một trong các class con của nó. Để chỉnh sửa các request thành công, bạn có thể override method serializeResponse, còn để chỉnh sửa các lỗi đi qua client này, bạn có thể override method serializeError. Để sử dụng custom class này, bạn truyền chính class đó vào method ClientsModule.register() thông qua property customClass. Dưới đây là ví dụ một custom ClientProxy serialize mỗi lỗi thành một RpcException.

import { ClientTCP, RpcException } from '@nestjs/microservices';

class ErrorHandlingProxy extends ClientTCP {
  serializeError(err: Error) {
    return new RpcException(err);
  }
}

và sau đó dùng nó trong ClientsModule như sau:

@Module({
  imports: [
    ClientsModule.register([{
      name: 'CustomProxy',
      customClass: ErrorHandlingProxy,
    }]),
  ]
})
export class AppModule