gRPC

gRPC là một framework RPC hiện đại, mã nguồn mở, hiệu năng cao, có thể chạy trong bất kỳ môi trường nào. Nó kết nối hiệu quả các service trong và giữa các data center, với khả năng cắm thêm (pluggable) hỗ trợ cho load balancing, tracing, health checking và authentication.

Giống như nhiều hệ thống RPC khác, gRPC dựa trên khái niệm định nghĩa một service dưới dạng các hàm (method) có thể được gọi từ xa. Với mỗi method, bạn định nghĩa tham số và kiểu trả về. Service, tham số và kiểu trả về được định nghĩa trong các file .proto, dùng cơ chế protocol buffers mã nguồn mở, không phụ thuộc ngôn ngữ (language-neutral) của Google.

Với gRPC transporter, Nest dùng file .proto để gắn (bind) client và server một cách động, giúp việc triển khai remote procedure call trở nên dễ dàng, tự động serialize và deserialize dữ liệu có cấu trúc.

Cài đặt

Để bắt đầu xây dựng microservice dùng gRPC, trước tiên cài các package cần thiết:

$ npm i --save @grpc/grpc-js @grpc/proto-loader

Tổng quan

Giống như các cách triển khai transport layer khác của Nest microservices, bạn chọn cơ chế gRPC transporter bằng property transport trong object options truyền vào method createMicroservice(). Trong ví dụ dưới đây, ta sẽ thiết lập một hero service. Property options cung cấp metadata về service đó; các property của nó được mô tả ở phần dưới.

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.GRPC,
  options: {
    package: 'hero',
    protoPath: join(import.meta.dirname, 'hero/hero.proto'),
  },
});

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.GRPC,
  options: {
    package: 'hero',
    protoPath: join(import.meta.dirname, 'hero/hero.proto'),
  },
});

Trong file nest-cli.json, ta thêm property assets để cho phép phân phối các file không phải TypeScript, và watchAssets để bật theo dõi (watch) tất cả các asset không phải TypeScript. Trong trường hợp này, ta muốn các file .proto được tự động copy vào thư mục dist.

{
  "compilerOptions": {
    "assets": ["**/*.proto"],
    "watchAssets": true
  }
}

Options

Object options của gRPC transporter cung cấp các property được mô tả dưới đây.

Ví dụ về gRPC service

Hãy định nghĩa một gRPC service mẫu tên là HeroesService. Trong object options ở trên, property protoPath thiết lập đường dẫn đến file định nghĩa .proto là hero.proto. File hero.proto được cấu trúc theo protocol buffers. Nội dung của nó như sau:

// hero/hero.proto
syntax = "proto3";

package hero;

service HeroesService {
  rpc FindOne (HeroById) returns (Hero) {}
}

message HeroById {
  int32 id = 1;
}

message Hero {
  int32 id = 1;
  string name = 2;
}

HeroesService của ta expose một method FindOne(). Method này nhận vào một tham số kiểu HeroById và trả về một message Hero (protocol buffers dùng phần tử message để định nghĩa cả kiểu tham số và kiểu trả về).

Tiếp theo, ta cần triển khai service này. Để định nghĩa một handler đáp ứng đúng định nghĩa trên, ta dùng decorator @GrpcMethod() trong một controller, như dưới đây. Decorator này cung cấp metadata cần thiết để khai báo một method là gRPC service method.

@Controller()
export class HeroesController {
  @GrpcMethod('HeroesService', 'FindOne')
  findOne(data: HeroById, metadata: Metadata, call: ServerUnaryCall<any, any>): Hero {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

@Controller()
export class HeroesController {
  @GrpcMethod('HeroesService', 'FindOne')
  findOne(data, metadata, call) {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

Decorator trên nhận hai tham số. Tham số đầu là tên service (ví dụ 'HeroesService'), tương ứng với định nghĩa service HeroesService trong hero.proto. Tham số thứ hai (chuỗi 'FindOne') tương ứng với rpc method FindOne() được định nghĩa bên trong HeroesService trong file hero.proto.

Method handler findOne() nhận ba tham số: data được truyền từ phía gọi, metadata lưu metadata của gRPC request, và call để lấy các property của object GrpcCall như sendMetadata dùng để gửi metadata về client.

Cả hai tham số của decorator @GrpcMethod() đều không bắt buộc. Nếu gọi mà không có tham số thứ hai (ví dụ 'FindOne'), Nest sẽ tự động liên kết rpc method trong file .proto với handler bằng cách chuyển tên handler sang dạng upper camel case (ví dụ handler findOne sẽ liên kết với định nghĩa rpc call FindOne). Điều này được minh họa dưới đây.

@Controller()
export class HeroesController {
  @GrpcMethod('HeroesService')
  findOne(data: HeroById, metadata: Metadata, call: ServerUnaryCall<any, any>): Hero {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

@Controller()
export class HeroesController {
  @GrpcMethod('HeroesService')
  findOne(data, metadata, call) {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

Bạn cũng có thể bỏ qua tham số đầu tiên của @GrpcMethod(). Trong trường hợp này, Nest sẽ tự động liên kết handler với định nghĩa service từ file định nghĩa proto dựa trên tên class nơi handler được định nghĩa. Ví dụ, trong đoạn code sau, class HeroesService liên kết các handler method của nó với định nghĩa service HeroesService trong file hero.proto dựa trên việc khớp tên 'HeroesService'.

@Controller()
export class HeroesService {
  @GrpcMethod()
  findOne(data: HeroById, metadata: Metadata, call: ServerUnaryCall<any, any>): Hero {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

@Controller()
export class HeroesService {
  @GrpcMethod()
  findOne(data, metadata, call) {
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];
    return items.find(({ id }) => id === data.id);
  }
}

Client

Các ứng dụng Nest có thể đóng vai trò là gRPC client, sử dụng (consume) các service được định nghĩa trong file .proto. Bạn truy cập các service từ xa thông qua một object ClientGrpc. Có nhiều cách để lấy được object ClientGrpc.

Cách được khuyến nghị là import ClientsModule. Dùng method register() để gắn (bind) một package các service được định nghĩa trong file .proto với một injection token, và cấu hình service đó. Property name chính là injection token. Với gRPC service, dùng transport: Transport.GRPC. Property options là một object có các property giống như đã mô tả ở trên.

imports: [
  ClientsModule.register([
    {
      name: 'HERO_PACKAGE',
      transport: Transport.GRPC,
      options: {
        package: 'hero',
        protoPath: join(import.meta.dirname, 'hero/hero.proto'),
      },
    },
  ]),
];

Sau khi đăng ký, ta có thể inject object ClientGrpc đã cấu hình bằng @Inject(). Sau đó dùng method getService() của object ClientGrpc để lấy instance của service, như dưới đây.

@Injectable()
export class AppService implements OnModuleInit {
  private heroesService: HeroesService;

  constructor(@Inject('HERO_PACKAGE') private client: ClientGrpc) {}

  onModuleInit() {
    this.heroesService = this.client.getService<HeroesService>('HeroesService');
  }

  getHero(): Observable<string> {
    return this.heroesService.findOne({ id: 1 });
  }
}

Lưu ý có một khác biệt nhỏ so với cách làm ở các microservice transport khác. Thay vì dùng class ClientProxy, ta dùng class ClientGrpc, class này cung cấp method getService(). Generic method getService() nhận vào tên service làm tham số và trả về instance tương ứng (nếu có).

Ngoài ra, bạn có thể dùng decorator @Client() để khởi tạo một object ClientGrpc, như sau:

@Injectable()
export class AppService implements OnModuleInit {
  @Client({
    transport: Transport.GRPC,
    options: {
      package: 'hero',
      protoPath: join(import.meta.dirname, 'hero/hero.proto'),
    },
  })
  client: ClientGrpc;

  private heroesService: HeroesService;

  onModuleInit() {
    this.heroesService = this.client.getService<HeroesService>('HeroesService');
  }

  getHero(): Observable<string> {
    return this.heroesService.findOne({ id: 1 });
  }
}

Cuối cùng, với các trường hợp phức tạp hơn, ta có thể inject một client được cấu hình động bằng class ClientProxyFactory như đã mô tả tại đây.

Dù dùng cách nào, ta cũng nhận được một tham chiếu đến proxy object HeroesService, object này expose cùng tập hợp các method được định nghĩa trong file .proto. Khi ta truy cập proxy object này (tức heroesService), hệ thống gRPC sẽ tự động serialize request, chuyển tiếp đến hệ thống từ xa, nhận response và deserialize response đó. Vì gRPC che giấu ta khỏi các chi tiết giao tiếp mạng này, heroesService trông và hoạt động giống như một provider cục bộ (local).

Lưu ý, tất cả các service method đều ở dạng lower camel case (để tuân theo quy ước tự nhiên của ngôn ngữ). Ví dụ, trong khi định nghĩa HeroesService ở file .proto chứa hàm FindOne(), thì instance heroesService sẽ cung cấp method findOne().

interface HeroesService {
  findOne(data: { id: number }): Observable<any>;
}

Một message handler cũng có thể trả về một Observable, trong trường hợp đó các giá trị kết quả sẽ được emit cho đến khi stream hoàn tất.

@Get()
call(): Observable<any> {
  return this.heroesService.findOne({ id: 1 });
}

@Get()
call() {
  return this.heroesService.findOne({ id: 1 });
}

Để gửi gRPC metadata (cùng với request), bạn có thể truyền vào tham số thứ hai, như sau:

call(): Observable<any> {
  const metadata = new Metadata();
  metadata.add('Set-Cookie', 'yummy_cookie=choco');

  return this.heroesService.findOne({ id: 1 }, metadata);
}

Lưu ý rằng điều này sẽ yêu cầu cập nhật lại interface HeroesService mà ta đã định nghĩa ở vài bước trước.

Xử lý exception (Exception handling)

Các gRPC handler có thể throw RpcException, nhưng một RpcException thông thường không mang theo gRPC status code, nên client sẽ nhận được UNKNOWN cho mọi lỗi. Kể từ NestJS v12, package microservices cung cấp sẵn các exception gRPC chuyên biệt và một GrpcExceptionFilter để map chúng thành các gRPC error object đúng chuẩn.

Throw một trong các exception theo status cụ thể từ handler của bạn:

import { GrpcAlreadyExistsException } from '@nestjs/microservices';

@GrpcMethod('HeroesService')
create(data: Hero): Hero {
  if (this.heroes.has(data.id)) {
    throw new GrpcAlreadyExistsException('Hero already exists');
  }
  return this.heroes.add(data);
}

Sau đó đăng ký filter để các exception này được serialize thành gRPC error:

import { GrpcExceptionFilter } from '@nestjs/microservices';

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.GRPC,
  options: {
    package: 'hero',
    protoPath: join(import.meta.dirname, 'hero/hero.proto'),
  },
});
app.useGlobalFilters(new GrpcExceptionFilter());

Khi đã có filter này, client sẽ nhận được ALREADY_EXISTS thay vì UNKNOWN.

Bạn cũng có thể dùng GrpcException dạng generic và truyền status một cách tường minh:

import { GrpcException, GrpcStatus } from '@nestjs/microservices';

throw new GrpcException('Rate limit exceeded', GrpcStatus.RESOURCE_EXHAUSTED);

Các class exception theo status cụ thể sau đây đều có sẵn, mỗi class tương ứng với một thành viên của enum GrpcStatus:

Ví dụ

Một ví dụ đầy đủ có thể xem tại đây.

gRPC Reflection

Đặc tả gRPC Server Reflection là một chuẩn cho phép gRPC client yêu cầu thông tin chi tiết về API mà server expose, tương tự như việc expose một tài liệu OpenAPI cho REST API. Điều này giúp việc làm việc với các công cụ debug như grpc-ui hay postman dễ dàng hơn đáng kể.

Để thêm hỗ trợ gRPC reflection vào server, trước tiên hãy cài package triển khai cần thiết:

$ npm i --save @grpc/reflection

Sau đó có thể gắn (hook) nó vào gRPC server bằng hook onLoadPackageDefinition trong options của gRPC server, như sau:

import { ReflectionService } from '@grpc/reflection';

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  options: {
    onLoadPackageDefinition: (pkg, server) => {
      new ReflectionService(pkg).addToServer(server);
    },
  },
});

Giờ đây server của bạn sẽ phản hồi các message yêu cầu thông tin API theo đặc tả reflection.

gRPC Streaming

Bản thân gRPC hỗ trợ các kết nối trực tiếp (live) dài hạn, thường được gọi là stream. Stream hữu ích cho các trường hợp như chat, theo dõi (observation) hoặc truyền dữ liệu theo từng phần (chunk-data). Xem thêm chi tiết trong tài liệu chính thức tại đây.

Nest hỗ trợ GRPC stream handler theo hai cách:

• Handler RxJS Subject + Observable: hữu ích khi cần viết response ngay trong method của Controller, hoặc truyền xuống cho consumer của Subject/Observable
• Handler GRPC call stream thuần túy: hữu ích khi cần truyền cho một executor nào đó sẽ đảm nhiệm phần còn lại của việc dispatch cho handler Duplex stream chuẩn của Node.

Ví dụ Streaming

Hãy định nghĩa một gRPC service mẫu mới tên là HelloService. File hello.proto được cấu trúc theo protocol buffers. Nội dung của nó như sau:

// hello/hello.proto
syntax = "proto3";

package hello;

service HelloService {
  rpc BidiHello(stream HelloRequest) returns (stream HelloResponse);
  rpc LotsOfGreetings(stream HelloRequest) returns (HelloResponse);
}

message HelloRequest {
  string greeting = 1;
}

message HelloResponse {
  string reply = 1;
}

Dựa trên file .proto này, hãy định nghĩa interface HelloService:

interface HelloService {
  bidiHello(upstream: Observable<HelloRequest>): Observable<HelloResponse>;
  lotsOfGreetings(
    upstream: Observable<HelloRequest>,
  ): Observable<HelloResponse>;
}

interface HelloRequest {
  greeting: string;
}

interface HelloResponse {
  reply: string;
}

Chiến lược Subject (Subject strategy)

Decorator @GrpcStreamMethod() cung cấp tham số hàm dưới dạng một Observable của RxJS. Nhờ đó, ta có thể nhận và xử lý nhiều message.

@GrpcStreamMethod()
bidiHello(messages: Observable<any>, metadata: Metadata, call: ServerDuplexStream<any, any>): Observable<any> {
  const subject = new Subject();

  const onNext = message => {
    console.log(message);
    subject.next({
      reply: 'Hello, world!'
    });
  };
  const onComplete = () => subject.complete();
  messages.subscribe({
    next: onNext,
    complete: onComplete,
  });

  return subject.asObservable();
}

Theo định nghĩa service (trong file .proto), method BidiHello sẽ stream request đến service. Để gửi nhiều message bất đồng bộ vào stream từ phía client, ta tận dụng class ReplaySubject của RxJS.

const helloService = this.client.getService<HelloService>('HelloService');
const helloRequest$ = new ReplaySubject<HelloRequest>();

helloRequest$.next({ greeting: 'Hello (1)!' });
helloRequest$.next({ greeting: 'Hello (2)!' });
helloRequest$.complete();

return helloService.bidiHello(helloRequest$);

Trong ví dụ trên, ta đã ghi hai message vào stream (các lệnh gọi next()) và báo cho service biết ta đã gửi xong dữ liệu (lệnh gọi complete()).

Call stream handler

Khi kiểu trả về của method được định nghĩa là stream, decorator @GrpcStreamCall() cung cấp tham số hàm dưới dạng grpc.ServerDuplexStream, hỗ trợ các method chuẩn như .on('data', callback), .write(message) hoặc .cancel(). Tài liệu đầy đủ về các method có sẵn có thể xem tại đây.

Ngược lại, khi kiểu trả về của method không phải là stream, decorator @GrpcStreamCall() cung cấp hai tham số hàm, lần lượt là grpc.ServerReadableStream (đọc thêm tại đây) và callback.

Hãy bắt đầu bằng việc triển khai BidiHello, method này cần hỗ trợ tương tác full-duplex.

@GrpcStreamCall()
bidiHello(requestStream: any) {
  requestStream.on('data', message => {
    console.log(message);
    requestStream.write({
      reply: 'Hello, world!'
    });
  });
}

Trong ví dụ trên, ta dùng method write() để ghi object vào response stream. Callback được truyền vào method .on() như tham số thứ hai sẽ được gọi mỗi khi service nhận được một chunk dữ liệu mới.

Hãy triển khai method LotsOfGreetings.

@GrpcStreamCall()
lotsOfGreetings(requestStream: any, callback: (err: unknown, value: HelloResponse) => void) {
  requestStream.on('data', message => {
    console.log(message);
  });
  requestStream.on('end', () => callback(null, { reply: 'Hello, world!' }));
}

Ở đây ta dùng hàm callback để gửi response sau khi việc xử lý requestStream đã hoàn tất.

Health check

Khi chạy một ứng dụng gRPC trong một orchestrator như Kubernetes, bạn có thể cần biết liệu nó có đang chạy và ở trạng thái healthy hay không. Đặc tả gRPC Health Check là một chuẩn cho phép gRPC client expose trạng thái health của mình để orchestrator xử lý phù hợp.

Để thêm hỗ trợ gRPC health check, trước tiên hãy cài package grpc-node:

$ npm i --save grpc-health-check

Sau đó có thể gắn nó vào gRPC service bằng hook onLoadPackageDefinition trong options của gRPC server, như sau. Lưu ý rằng protoPath cần có cả package health check lẫn package hero.

import { HealthImplementation, protoPath as healthCheckProtoPath } from 'grpc-health-check';

const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  options: {
    protoPath: [
      healthCheckProtoPath,
      protoPath: join(import.meta.dirname, 'hero/hero.proto'),
    ],
    onLoadPackageDefinition: (pkg, server) => {
      const healthImpl = new HealthImplementation({
        '': 'UNKNOWN',
      });

      healthImpl.addToServer(server);
      healthImpl.setStatus('', 'SERVING');
    },
  },
});

gRPC Metadata

Metadata là thông tin về một RPC call cụ thể, ở dạng danh sách các cặp key-value, trong đó key là string còn value thường là string nhưng cũng có thể là dữ liệu binary. Metadata trong suốt (opaque) đối với bản thân gRPC - nó cho phép client cung cấp thông tin liên quan đến call cho server và ngược lại. Metadata có thể bao gồm authentication token, request identifier và tag phục vụ mục đích giám sát (monitoring), cũng như thông tin dữ liệu như số lượng bản ghi trong một tập dữ liệu.

Để đọc metadata trong handler @GrpcMethod(), dùng tham số thứ hai (metadata), có kiểu Metadata (import từ package grpc).

Để gửi metadata trở lại từ handler, dùng method ServerUnaryCall#sendMetadata() (tham số thứ ba của handler).

@Controller()
export class HeroesService {
  @GrpcMethod()
  findOne(data: HeroById, metadata: Metadata, call: ServerUnaryCall<any, any>): Hero {
    const serverMetadata = new Metadata();
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];

    serverMetadata.add('Set-Cookie', 'yummy_cookie=choco');
    call.sendMetadata(serverMetadata);

    return items.find(({ id }) => id === data.id);
  }
}

@Controller()
export class HeroesService {
  @GrpcMethod()
  findOne(data, metadata, call) {
    const serverMetadata = new Metadata();
    const items = [
      { id: 1, name: 'John' },
      { id: 2, name: 'Doe' },
    ];

    serverMetadata.add('Set-Cookie', 'yummy_cookie=choco');
    call.sendMetadata(serverMetadata);

    return items.find(({ id }) => id === data.id);
  }
}

Tương tự, để đọc metadata trong các handler được đánh dấu bằng @GrpcStreamMethod() (chiến lược Subject), dùng tham số thứ hai (metadata), có kiểu Metadata (import từ package grpc).

Để gửi metadata trở lại từ handler, dùng method ServerDuplexStream#sendMetadata() (tham số thứ ba của handler).

Để đọc metadata từ bên trong các call stream handler (handler được đánh dấu bằng decorator @GrpcStreamCall()), lắng nghe event metadata trên tham chiếu requestStream, như sau:

requestStream.on('metadata', (metadata: Metadata) => {
  const meta = metadata.get('X-Meta');
});