GRPCの非同期処理をPromiseで制御する

GRPCの通信部分をPromiseを使って可読性を意識しながら書いた記事になります。
※GRPC通信に使用しているのはimprobable-eng/grpc-webです

概要

以下はGRPC-webを使用してUserの取得、更新、削除を行うサービスのStore部分です。

store/index.js

import { grpc } from 'grpc-web-client';
import { DeleteUserRequest, GetUserRequest, SetUserRequest, User } from '../ts/_proto/src/user_pb';
import { UserService } from '../ts/_proto/src/user_pb_service';

const defaultUser = {
  name: 'hoge',
  followerIdsList: [],
  info: {
    description: '',
  },
};

export const state = () => ({
  user: defaultUser,
});

export const mutations = {
  setUser(state, { user }) {
    state.user = user;
  },
  resetUser(state) {
    state.user = defaultUser;
  },
};

export const actions = {
  getUser({ commit }, { id }) {
    const client = grpc.client(UserService.GetUser, {
      host: 'http://localhost:9090',
    });
    const param = new GetUserRequest();
    param.setId(id);
    client.start();
    client.send(param);
    client.onMessage(message => {
      const pbUser = message.getUser();
      const user = pbUser.toObject();
      commit('setUser', {
        user,
      });
    });
  },
  setUser({ commit, state }, { id }) {
    const pbInfo = new User.Info();
    pbInfo.setDescription(state.user.info.description);

    const pbUser = new User();
    pbUser.setName(state.user.name);
    pbUser.setFollowerIdsList(state.user.followerIdsList);
    pbUser.setInfo(pbInfo);

    const param = new SetUserRequest();
    param.setId(id);
    param.setUser(pbUser);

    const client = grpc.client(UserService.SetUser, {
      host: 'http://localhost:9090',
    });
    client.start();
    client.send(param);
    client.onMessage(message => {
      const pbUser = message.getUser();
      const user = pbUser.toObject();
      commit('setUser', {
        user,
      });
    });
  },
  deleteUser({ commit }, { id }) {
    const param = new DeleteUserRequest();
    param.setId(id);

    const client = grpc.client(UserService.DeleteUser, {
      host: 'http://localhost:9090',
    });
    client.start();
    client.send(param);
    client.onMessage(message => {
      if (message.getResult()) {
        commit('resetUser');
      };
    });
  },
};

全てのActionは通信を行ってその結果をcommitするのみなのですが、上記のようにやっていることは単純でも通信部分のコードは行数が多く冗長になってしまいます。

解決策と懸念点

そこでclient~onMessageのGRPC通信部分を関数に切り出すことにしました。しかし１つある懸念点として、onMessageは非同期で行われるので

    client.onMessage(message => {
      return message
    });

と書くことはできません。またonMessage後にやりたい処理をcallback関数にして通信部分として切り出した関数に渡してあげるのも微妙な感じがします。
そこで切り出した関数部分はPromiseを返すように書きます。

index.ts

// 中略

export const actions = {
  async getUser({ commit }, { id }) {
    const res = await getUserRPC(id);
    const pbUser = res.getUser();
    const user = pbUser.toObject();
    commit('setUser', {
      user,
    });
  },
  async setUser({ commit, state }, { id }) {
    const res = await setUserRPC(id, state.user);
    const pbUser = res.getUser();
    const user = pbUser.toObject();
    commit('setUser', {
      user,
    });
  },
  async deleteUser({ commit }, { id }) {
    const res = await deleteUserRPC(id);
    if (res.getResult()) {
      commit('resetUser')
    }
  },
};

// ここからが切り出したGRPC通信部分

const getUserRPC = id => {
  return new Promise(resolve => {
    const client = grpc.client(UserService.GetUser, {
      host: 'http://localhost:9090',
    });
    const param = new GetUserRequest();
    param.setId(id);
    client.start();
    client.send(param);
    client.onMessage(message => {
      resolve(message);
    });
  });
}

const setUserRPC = (id, user) => {
  return new Promise((resolve, reject) => {
    const pbInfo = new User.Info();
    pbInfo.setDescription(user.info.description);

    const pbUser = new User();
    pbUser.setName(user.name);
    pbUser.setFollowerIdsList(user.followerIdsList);
    pbUser.setInfo(pbInfo);

    const param = new SetUserRequest();
    param.setId(id);
    param.setUser(pbUser);

    const client = grpc.client(UserService.SetUser, {
      host: 'http://localhost:9090',
    });
    client.start();
    client.send(param);
    client.onMessage(message => {
      resolve(message);
    });
  });
}

const deleteUserRPC = id => {
  return new Promise((resolve) => {
    const param = new DeleteUserRequest();
    param.setId(id);

    const client = grpc.client(UserService.DeleteUser, {
      host: 'http://localhost:9090',
    });
    client.start();
    client.send(param);
    client.onMessage(message => {
      resolve(message);
    });
  });
}

const res = await...の部分で切り出した関数の中でresolveされるまで待機されるので、通信処理が意図したところまで終わった時点で初めてconst resに値が入るようになります。上記の書くことでActionsの行数が大幅に削減されますし、また値の戻り値をそのまま使えるという点でonMessageによる非同期処理の複雑さをActionsは意識せず直感的に理解できるコードになります。

複数の通信に対応する

今度は１度に複数の通信を行う場合を考えてみたいと思います。
Storeの初期化処理でユーザーの情報と、そのユーザーの設定に関する情報を取得します。

store/index.ts

import { grpc } from 'grpc-web-client';
import { GetUserRequest, GetUserConfigRequest } from '../ts/_proto/src/user_pb';
import { UserService } from '../ts/_proto/src/user_pb_service';

const defaultUser = {
  name: 'hoge',
  followerIdsList: [],
  info: {
    description: '',
  },
};

const defaultUserConfig = {
  isPublic: false,
  isOfficial: false
}

export const state = () => ({
  user: defaultUser,
  config: defaultUserConfig
});

export const mutations = {
  setUser(state, { user }) {
    state.user = user;
  },
  setConfig(state, { config }) {
    state.config = config;
  },
};

export const actions = {
  async init({ commit }, { id }) {
    const getUserResult = await getUserRPC(id);
    const pbUser = getUserResult.getUser();
    const user = pbUser.toObject();
    commit('setUser', {
      user,
    });

    const getUserConfigResult = await getUserConfigRPC(id);
    const pbUserConfig = getUserConfigResult.getConfig();
    const config = pbUserConfig.toObject();
    commit('setConfig', {
      config,
    });
  },
};

const getUserRPC = id => {
  return new Promise(resolve => {
    const client = grpc.client(UserService.GetUser, {
      host: 'http://localhost:9090',
    });
    const param = new GetUserRequest();
    param.setId(id);
    client.start();
    client.send(param);
    client.onMessage(message => {
      resolve(message);
    });
  });
}

const getUserConfigRPC = id => {
  return new Promise(resolve => {
    const client = grpc.client(UserService.GetUserConfig, {
      host: 'http://localhost:9090',
    });
    const param = new GetUserConfigRequest();
    param.setId(id);
    client.start();
    client.send(param);
    client.onMessage(message => {
      resolve(message);
    });
  });
}

前半に紹介したやり方でユーザーの情報とユーザーの設定に関する情報を取得しましたが、この場合問題になるのがこの２つの処理が直列で行われているというところです。複数の処理を１つ１つ待つのはあまりスマートではないので、並列で実行して全ての通信処理が終わったら次の処理を行うというようにしたいです。

これをPromise.allを使って以下のように書くことができます。

store/index.js

// 中略

export const actions = {
  async init({ commit }, { id }) {
    Promise.all([await getUserRPC(id), await getUserConfigRPC(id)]).then(values => {
      const getUserResult = values[0];
      const pbUser = getUserResult.getUser();
      const user = pbUser.toObject();
      commit('setUser', {
        user,
      });

      const getUserConfigResult = values[1];
      const pbUserConfig = getUserConfigResult.getConfig();
      const config = pbUserConfig.toObject();
      commit('setConfig', {
        config,
      });
    }).catch(e => {
      // error handling
    });
  },
};

// 中略

Promise.allで第一引数の配列に各Promiseを返す関数を入れることでその全てがresolveされてからthenに入るので並列で実行して終わってから値の取得とstateへのcommitを行うことができます。各値の取得からcommitまでの部分も関数に切り出せばより可読性が高いコードになりそうです。

上記のようにPromiseを使うことでActionsのGRPC通信の可読性の向上と複数通信を並列実行を実現できました。また通信部分を切り出した部分をまとめたものを別ファイルにおいておけば通信部分のみをテストすることも可能になるので、そういった部分に関しても恩恵を受けることができそうです。