Subscriptions

Bên cạnh việc lấy dữ liệu bằng query và sửa dữ liệu bằng mutation, đặc tả GraphQL còn hỗ trợ một loại operation thứ ba, gọi là subscription. GraphQL subscription là cách để đẩy dữ liệu từ server tới các client chọn lắng nghe (listen) các message thời gian thực từ server. Subscription tương tự query ở chỗ chúng cũng chỉ định một tập field cần trả về cho client, nhưng thay vì trả về ngay một kết quả duy nhất, một kênh (channel) sẽ được mở ra và một kết quả sẽ được gửi cho client mỗi khi một sự kiện cụ thể xảy ra trên server.

Một trường hợp sử dụng phổ biến của subscription là thông báo cho phía client về các sự kiện cụ thể, ví dụ việc tạo một object mới, các field được cập nhật, v.v. (đọc thêm tại đây).

Bật subscription với driver Apollo

Subscription được truyền tải qua WebSocket bởi package graphql-ws, package này không được đóng gói kèm theo @nestjs/apollo. Cài nó trước:

$ npm i --save graphql-ws

Sau đó bật transport một cách tường minh:

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  subscriptions: {
    'graphql-ws': true,
  },
}),

Code first

Để tạo một subscription bằng cách tiếp cận code first, chúng ta dùng decorator @Subscription() (được export từ package @nestjs/graphql) và class PubSub từ package graphql-subscriptions, class này cung cấp một API publish/subscribe đơn giản.

Subscription handler dưới đây đảm nhiệm việc đăng ký lắng nghe (subscribing) một sự kiện bằng cách gọi PubSub#asyncIterableIterator. Phương thức này nhận một tham số duy nhất, triggerName, tương ứng với tên topic của sự kiện.

const pubSub = new PubSub();

@Resolver(() => Author)
export class AuthorResolver {
  // ...
  @Subscription(() => Comment)
  commentAdded() {
    return pubSub.asyncIterableIterator('commentAdded');
  }
}

Đoạn code trên sẽ sinh ra phần schema GraphQL sau ở dạng SDL:

type Subscription {
  commentAdded(): Comment!
}

Lưu ý rằng theo định nghĩa, subscription trả về một object với một thuộc tính cấp cao nhất (top level) duy nhất, có key là tên của subscription. Tên này hoặc được kế thừa từ tên của phương thức subscription handler (tức commentAdded ở trên), hoặc được cung cấp tường minh bằng cách truyền một option với key name làm tham số thứ hai cho decorator @Subscription(), như minh họa bên dưới.

@Subscription(() => Comment, {
  name: 'commentAdded',
})
subscribeToCommentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Cách viết này sinh ra cùng một SDL như ví dụ code trước, nhưng cho phép chúng ta tách rời tên phương thức khỏi tên subscription.

Publish (phát) sự kiện

Bây giờ, để publish sự kiện, chúng ta dùng phương thức PubSub#publish. Phương thức này thường được dùng bên trong một mutation để kích hoạt một cập nhật phía client khi một phần của object graph đã thay đổi. Ví dụ:

@Mutation(() => Comment)
async addComment(
  @Args('postId', { type: () => Int }) postId: number,
  @Args('comment', { type: () => Comment }) comment: CommentInput,
) {
  const newComment = this.commentsService.addComment({ id: postId, comment });
  pubSub.publish('commentAdded', { commentAdded: newComment });
  return newComment;
}

Phương thức PubSub#publish nhận một triggerName (một lần nữa, hãy nghĩ đây là tên topic của sự kiện) làm tham số đầu tiên, và một event payload làm tham số thứ hai. Như đã đề cập, theo định nghĩa, subscription trả về một giá trị, và giá trị đó có một hình dạng (shape) cụ thể. Hãy nhìn lại SDL được sinh ra cho subscription commentAdded của chúng ta:

type Subscription {
  commentAdded(): Comment!
}

Điều này cho chúng ta biết rằng subscription phải trả về một object với tên thuộc tính cấp cao nhất là commentAdded, có giá trị là một object Comment. Điểm quan trọng cần lưu ý là hình dạng của event payload do phương thức PubSub#publish phát ra phải tương ứng với hình dạng của giá trị mà subscription được mong đợi trả về. Vì vậy, trong ví dụ trên, câu lệnh pubSub.publish('commentAdded', { commentAdded: newComment }) publish một sự kiện commentAdded với payload có hình dạng phù hợp. Nếu các hình dạng này không khớp nhau, subscription của bạn sẽ thất bại trong giai đoạn kiểm tra hợp lệ (validation) của GraphQL.

Lọc subscription (Filtering subscriptions)

Để lọc bỏ các sự kiện cụ thể, đặt thuộc tính filter thành một hàm lọc (filter function). Hàm này hoạt động tương tự hàm được truyền cho filter của array. Nó nhận hai tham số: payload chứa event payload (do bên publish sự kiện gửi), và variables chứa bất kỳ tham số nào được truyền vào trong request subscription. Nó trả về một boolean xác định liệu sự kiện này có nên được publish tới các client đang lắng nghe hay không.

@Subscription(() => Comment, {
  filter: (payload, variables) =>
    payload.commentAdded.title === variables.title,
})
commentAdded(@Args('title') title: string) {
  return pubSub.asyncIterableIterator('commentAdded');
}

Biến đổi payload của subscription (Mutating subscription payloads)

Để biến đổi (mutate) event payload trước khi publish, đặt thuộc tính resolve thành một hàm. Hàm này nhận event payload (do bên publish sự kiện gửi) và trả về giá trị phù hợp.

@Subscription(() => Comment, {
  resolve: value => value,
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Nếu bạn cần truy cập các provider đã được inject (ví dụ, dùng một service bên ngoài để kiểm tra tính hợp lệ của dữ liệu), hãy dùng cấu trúc sau.

@Subscription(() => Comment, {
  resolve(this: AuthorResolver, value) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return value;
  }
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Cấu trúc tương tự cũng áp dụng được với filter:

@Subscription(() => Comment, {
  filter(this: AuthorResolver, payload, variables) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return payload.commentAdded.title === variables.title;
  }
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Schema first

Để tạo một subscription tương đương trong Nest, chúng ta sẽ dùng decorator @Subscription().

const pubSub = new PubSub();

@Resolver('Author')
export class AuthorResolver {
  // ...
  @Subscription()
  commentAdded() {
    return pubSub.asyncIterableIterator('commentAdded');
  }
}

Để lọc bỏ các sự kiện cụ thể dựa trên context và các tham số, đặt thuộc tính filter.

@Subscription('commentAdded', {
  filter: (payload, variables) =>
    payload.commentAdded.title === variables.title,
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Để biến đổi payload trước khi publish, chúng ta có thể dùng một hàm resolve.

@Subscription('commentAdded', {
  resolve: value => value,
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Nếu bạn cần truy cập các provider đã được inject (ví dụ, dùng một service bên ngoài để kiểm tra tính hợp lệ của dữ liệu), hãy dùng cấu trúc sau:

@Subscription('commentAdded', {
  resolve(this: AuthorResolver, value) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return value;
  }
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Cấu trúc tương tự cũng áp dụng được với filter:

@Subscription('commentAdded', {
  filter(this: AuthorResolver, payload, variables) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return payload.commentAdded.title === variables.title;
  }
})
commentAdded() {
  return pubSub.asyncIterableIterator('commentAdded');
}

Bước cuối cùng là cập nhật file định nghĩa type.

type Author {
  id: Int!
  firstName: String
  lastName: String
  posts: [Post]
}

type Post {
  id: Int!
  title: String
  votes: Int
}

type Query {
  author(id: Int!): Author
}

type Comment {
  id: String
  content: String
}

type Subscription {
  commentAdded(title: String!): Comment
}

Với điều này, chúng ta đã tạo ra một subscription duy nhất commentAdded(title: String!): Comment. Bạn có thể tìm một ví dụ implementation đầy đủ tại đây.

PubSub

Chúng ta đã khởi tạo một instance PubSub cục bộ (local) ở trên. Cách tiếp cận được khuyến nghị là định nghĩa PubSub như một provider và inject nó thông qua constructor (dùng decorator @Inject()). Điều này cho phép chúng ta tái sử dụng cùng một instance trong toàn bộ ứng dụng. Ví dụ, định nghĩa một provider như sau, sau đó inject 'PUB_SUB' ở nơi cần dùng.

{
  provide: 'PUB_SUB',
  useValue: new PubSub(),
}

Tùy chỉnh subscriptions server

Để tùy chỉnh subscriptions server (ví dụ, thay đổi path), dùng thuộc tính option subscriptions với graphql-ws:

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  subscriptions: {
    'graphql-ws': {
      path: '/graphql',
    },
  },
});

Xác thực qua WebSocket (Authentication over WebSockets)

Việc kiểm tra xem người dùng đã được xác thực hay chưa có thể thực hiện bên trong hàm callback onConnect mà bạn có thể chỉ định trong option subscriptions.

Với graphql-ws, callback onConnect nhận vào connection context, bao gồm cả connectionParams.

GraphQLModule.forRoot<ApolloDriverConfig>({
  driver: ApolloDriver,
  subscriptions: {
    'graphql-ws': {
      onConnect: (context: Context<any>) => {
        const { connectionParams, extra } = context;
        const authToken = connectionParams.authToken;
        if (!isValid(authToken)) {
          throw new Error('Token is not valid');
        }
        // khi dùng graphql-ws, các giá trị context bổ sung nên được lưu trong field extra
        extra.user = { user: {} };
      },
    },
  },
  context: ({ extra }) => {
    // giờ bạn có thể truy cập giá trị context bổ sung của mình thông qua field extra
  },
});

Bật subscription với driver Mercurius

Để bật subscription, đặt thuộc tính subscription thành true.

GraphQLModule.forRoot<MercuriusDriverConfig>({
  driver: MercuriusDriver,
  subscription: true,
}),

Code first

Để tạo một subscription bằng cách tiếp cận code first, chúng ta dùng decorator @Subscription() (được export từ package @nestjs/graphql) và class PubSub từ package mercurius, class này cung cấp một API publish/subscribe đơn giản.

Subscription handler dưới đây đảm nhiệm việc đăng ký lắng nghe (subscribing) một sự kiện bằng cách gọi PubSub#asyncIterableIterator. Phương thức này nhận một tham số duy nhất, triggerName, tương ứng với tên topic của sự kiện.

@Resolver(() => Author)
export class AuthorResolver {
  // ...
  @Subscription(() => Comment)
  commentAdded(@Context('pubsub') pubSub: PubSub) {
    return pubSub.subscribe('commentAdded');
  }
}

Đoạn code trên sẽ sinh ra phần schema GraphQL sau ở dạng SDL:

type Subscription {
  commentAdded(): Comment!
}

Lưu ý rằng theo định nghĩa, subscription trả về một object với một thuộc tính cấp cao nhất duy nhất, có key là tên của subscription. Tên này hoặc được kế thừa từ tên của phương thức subscription handler (tức commentAdded ở trên), hoặc được cung cấp tường minh bằng cách truyền một option với key name làm tham số thứ hai cho decorator @Subscription(), như minh họa bên dưới.

@Subscription(() => Comment, {
  name: 'commentAdded',
})
subscribeToCommentAdded(@Context('pubsub') pubSub: PubSub) {
  return pubSub.subscribe('commentAdded');
}

Cách viết này sinh ra cùng một SDL như ví dụ code trước, nhưng cho phép chúng ta tách rời tên phương thức khỏi tên subscription.

Publish (phát) sự kiện

Bây giờ, để publish sự kiện, chúng ta dùng phương thức PubSub#publish. Phương thức này thường được dùng bên trong một mutation để kích hoạt một cập nhật phía client khi một phần của object graph đã thay đổi. Ví dụ:

@Mutation(() => Comment)
async addComment(
  @Args('postId', { type: () => Int }) postId: number,
  @Args('comment', { type: () => Comment }) comment: CommentInput,
  @Context('pubsub') pubSub: PubSub,
) {
  const newComment = this.commentsService.addComment({ id: postId, comment });
  await pubSub.publish({
    topic: 'commentAdded',
    payload: {
      commentAdded: newComment
    }
  });
  return newComment;
}

Như đã đề cập, theo định nghĩa, subscription trả về một giá trị, và giá trị đó có một hình dạng cụ thể. Hãy nhìn lại SDL được sinh ra cho subscription commentAdded của chúng ta:

type Subscription {
  commentAdded(): Comment!
}

Điều này cho chúng ta biết rằng subscription phải trả về một object với tên thuộc tính cấp cao nhất là commentAdded, có giá trị là một object Comment. Điểm quan trọng cần lưu ý là hình dạng của event payload do phương thức PubSub#publish phát ra phải tương ứng với hình dạng của giá trị mà subscription được mong đợi trả về. Vì vậy, trong ví dụ trên, câu lệnh pubSub.publish({ topic: 'commentAdded', payload: { commentAdded: newComment } }) publish một sự kiện commentAdded với payload có hình dạng phù hợp. Nếu các hình dạng này không khớp nhau, subscription của bạn sẽ thất bại trong giai đoạn kiểm tra hợp lệ của GraphQL.

Lọc subscription (Filtering subscriptions)

Để lọc bỏ các sự kiện cụ thể, đặt thuộc tính filter thành một hàm lọc. Hàm này hoạt động tương tự hàm được truyền cho filter của array. Nó nhận hai tham số: payload chứa event payload (do bên publish sự kiện gửi), và variables chứa bất kỳ tham số nào được truyền vào trong request subscription. Nó trả về một boolean xác định liệu sự kiện này có nên được publish tới các client đang lắng nghe hay không.

@Subscription(() => Comment, {
  filter: (payload, variables) =>
    payload.commentAdded.title === variables.title,
})
commentAdded(@Args('title') title: string, @Context('pubsub') pubSub: PubSub) {
  return pubSub.subscribe('commentAdded');
}

Nếu bạn cần truy cập các provider đã được inject (ví dụ, dùng một service bên ngoài để kiểm tra tính hợp lệ của dữ liệu), hãy dùng cấu trúc sau.

@Subscription(() => Comment, {
  filter(this: AuthorResolver, payload, variables) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return payload.commentAdded.title === variables.title;
  }
})
commentAdded(@Args('title') title: string, @Context('pubsub') pubSub: PubSub) {
  return pubSub.subscribe('commentAdded');
}

Schema first

Để tạo một subscription tương đương trong Nest, chúng ta sẽ dùng decorator @Subscription().

const pubSub = new PubSub();

@Resolver('Author')
export class AuthorResolver {
  // ...
  @Subscription()
  commentAdded(@Context('pubsub') pubSub: PubSub) {
    return pubSub.subscribe('commentAdded');
  }
}

Để lọc bỏ các sự kiện cụ thể dựa trên context và các tham số, đặt thuộc tính filter.

@Subscription('commentAdded', {
  filter: (payload, variables) =>
    payload.commentAdded.title === variables.title,
})
commentAdded(@Context('pubsub') pubSub: PubSub) {
  return pubSub.subscribe('commentAdded');
}

Nếu bạn cần truy cập các provider đã được inject (ví dụ, dùng một service bên ngoài để kiểm tra tính hợp lệ của dữ liệu), hãy dùng cấu trúc sau:

@Subscription('commentAdded', {
  filter(this: AuthorResolver, payload, variables) {
    // "this" tham chiếu tới một instance của "AuthorResolver"
    return payload.commentAdded.title === variables.title;
  }
})
commentAdded(@Context('pubsub') pubSub: PubSub) {
  return pubSub.subscribe('commentAdded');
}

Bước cuối cùng là cập nhật file định nghĩa type.

type Author {
  id: Int!
  firstName: String
  lastName: String
  posts: [Post]
}

type Post {
  id: Int!
  title: String
  votes: Int
}

type Query {
  author(id: Int!): Author
}

type Comment {
  id: String
  content: String
}

type Subscription {
  commentAdded(title: String!): Comment
}

Với điều này, chúng ta đã tạo ra một subscription duy nhất commentAdded(title: String!): Comment.

PubSub

Trong các ví dụ trên, chúng ta đã dùng emitter PubSub mặc định (mqemitter). Cách tiếp cận được khuyến nghị (cho production) là dùng mqemitter-redis. Ngoài ra, bạn cũng có thể cung cấp một implementation PubSub tùy chỉnh (đọc thêm tại đây).

GraphQLModule.forRoot<MercuriusDriverConfig>({
  driver: MercuriusDriver,
  subscription: {
    emitter: require('mqemitter-redis')({
      port: 6579,
      host: '127.0.0.1',
    }),
  },
});

Xác thực qua WebSocket (Authentication over WebSockets)

Việc kiểm tra xem người dùng đã được xác thực hay chưa có thể thực hiện bên trong hàm callback verifyClient mà bạn có thể chỉ định trong option subscription.

verifyClient sẽ nhận object info làm tham số đầu tiên, bạn có thể dùng nó để lấy header của request.

GraphQLModule.forRoot<MercuriusDriverConfig>({
  driver: MercuriusDriver,
  subscription: {
    verifyClient: (info, next) => {
      const authorization = info.req.headers?.authorization as string;
      if (!authorization?.startsWith('Bearer ')) {
        return next(false);
      }
      next(true);
    },
  }
}),