test(grpc): add case base-app-stream-meta

packages/grpc/test/fixtures/base-app-stream-meta/package.json

@@ -0,0 +1,3 @@
+{
+  "name": "ali-demo"
+}

packages/grpc/test/fixtures/base-app-stream-meta/src/config/config.default.ts

@@ -0,0 +1,11 @@
+import { join } from 'path';
+
+export const grpcServer = {
+  url: 'localhost:6571',
+  services: [
+    {
+      protoPath: join(__dirname, '../../../proto/math.proto'),
+      package: 'math',
+    }
+  ],
+}

packages/grpc/test/fixtures/base-app-stream-meta/src/configuration.ts

@@ -0,0 +1,14 @@
+import { Configuration } from '@midwayjs/core';
+import * as grpc from '../../../../src';
+import { join } from 'path';
+
+@Configuration({
+  imports: [
+    grpc
+  ],
+  importConfigs: [
+    join(__dirname, './config'),
+  ]
+})
+export class AutoConfiguration {
+}

packages/grpc/test/fixtures/base-app-stream-meta/src/interface.ts

@@ -0,0 +1,41 @@
+import {
+  IClientDuplexStreamService,
+  IClientReadableStreamService,
+  IClientUnaryService,
+  IClientWritableStreamService
+} from '../../../../src';
+
+export namespace math {
+  export interface AddArgs {
+    id?: number;
+    num?: number;
+  }
+  export interface Num {
+    id?: number;
+    num?: number;
+  }
+
+  /**
+   * server interface
+   */
+  export interface Math {
+    add(data: AddArgs): Promise<Num>;
+    addMore(data: AddArgs): Promise<void>;
+    // 服务端推，客户端读
+    sumMany(fibArgs: AddArgs): Promise<void>
+    // 客户端端推，服务端读
+    addMany(num: AddArgs): Promise<void>;
+  }
+
+  /**
+   * client interface
+   */
+  export interface MathClient {
+    add(): IClientUnaryService<AddArgs, Num>;
+    addMore(): IClientDuplexStreamService<AddArgs, Num>;
+    // 服务端推，客户端读
+    sumMany(): IClientReadableStreamService<AddArgs, Num>;
+    // 客户端端推，服务端读
+    addMany(): IClientWritableStreamService<AddArgs, Num>;
+  }
+}

packages/grpc/test/fixtures/base-app-stream-meta/src/provider/math.ts

@@ -0,0 +1,119 @@
+import * as assert from 'assert';
+import { GrpcMethod, GrpcStreamTypeEnum, Inject, MSProviderType, Provide, Provider } from '@midwayjs/core';
+import { Context } from '../../../../../src';
+import { math } from '../interface';
+import { Metadata } from '@grpc/grpc-js';
+
+/**
+ */
+@Provide()
+@Provider(MSProviderType.GRPC, { package: 'math' })
+export class Math implements math.Math {
+
+  @Inject()
+  ctx: Context;
+
+  sumDataList = [];
+
+  @GrpcMethod()
+  async add(data: math.AddArgs): Promise<math.Num> {
+    assert(this.ctx, 'should get context');
+    const { metadata } = this.ctx;
+    assert(metadata, 'should get metadata');
+
+    const rpcDefinition = metadata.get('rpc.definition');
+    assert(rpcDefinition[0] === 'math.Math', `should get rpc.definition, but got "${rpcDefinition}"`);
+
+    const rpcMethod = metadata.get('rpc.method');
+    assert(rpcMethod[0] === 'Add', `should get rpc.method, but got "${rpcMethod[0]}"`);
+
+    const rpcMethodType = metadata.get('rpc.method.type');
+    assert(rpcMethodType[0] === 'unary', `should get rpc.method.type, but got "${rpcMethodType[0]}"`);
+
+    return {
+      num: data.num + 2,
+    }
+  }
+
+  @GrpcMethod({type: GrpcStreamTypeEnum.DUPLEX, onEnd: 'duplexEnd' })
+  async addMore(message: math.AddArgs)  {
+    const { metadata } = this.ctx;
+    assert(metadata, 'should get metadata');
+
+    const rpcDefinition = metadata.get('rpc.definition');
+    assert(rpcDefinition[0] === 'math.Math', `should get rpc.definition, but got "${rpcDefinition}"`);
+
+    const rpcMethod = metadata.get('rpc.method');
+    assert(rpcMethod[0] === 'AddMore', `should get rpc.method, but got "${rpcMethod[0]}"`);
+
+    const rpcMethodType = metadata.get('rpc.method.type');
+    assert(rpcMethodType[0] === 'bidi', `should get rpc.method.type, but got "${rpcMethodType[0]}"`);
+
+    this.ctx.write({
+      id: message.id,
+      num: message.num + 10,
+    });
+  }
+
+  async duplexEnd() {
+    console.log('got client end message');
+  }
+
+  @GrpcMethod({type: GrpcStreamTypeEnum.WRITEABLE })
+  async sumMany(args: math.AddArgs) {
+    const { metadata } = this.ctx;
+    assert(metadata, 'should get metadata');
+
+    const rpcDefinition = metadata.get('rpc.definition');
+    assert(rpcDefinition[0] === 'math.Math', `should get rpc.definition, but got "${rpcDefinition}"`);
+
+    const rpcMethod = metadata.get('rpc.method');
+    assert(rpcMethod[0] === 'SumMany', `should get rpc.method, but got "${rpcMethod[0]}"`);
+
+    const rpcMethodType = metadata.get('rpc.method.type');
+    assert(rpcMethodType[0] === 'server', `should get rpc.method.type, but got "${rpcMethodType[0]}"`);
+
+    this.ctx.write({
+      num: 1 + args.num
+    });
+    this.ctx.write({
+      num: 2 + args.num
+    });
+    this.ctx.write({
+      num: 3 + args.num
+    });
+
+    const meta = new Metadata();
+    meta.add('xxx', 'bbb');
+
+    this.ctx.sendMetadata(meta);
+    this.ctx.end();
+  }
+
+  @GrpcMethod({type: GrpcStreamTypeEnum.READABLE, onEnd: 'sumEnd' })
+  async addMany(data: math.Num) {
+    const { metadata } = this.ctx;
+    assert(metadata, 'should get metadata');
+
+    const rpcDefinition = metadata.get('rpc.definition');
+    assert(rpcDefinition[0] === 'math.Math', `should get rpc.definition, but got "${rpcDefinition}"`);
+
+    const rpcMethod = metadata.get('rpc.method');
+    assert(rpcMethod[0] === 'AddMany', `should get rpc.method, but got "${rpcMethod[0]}"`);
+
+    const rpcMethodType = metadata.get('rpc.method.type');
+    assert(rpcMethodType[0] === 'client', `should get rpc.method.type, but got "${rpcMethodType[0]}"`);
+
+    this.sumDataList.push(data);
+  }
+
+  async sumEnd(): Promise<math.Num> {
+    const total = this.sumDataList.reduce((pre, cur) => {
+      return {
+        num: pre.num + cur.num,
+      }
+    });
+    return total;
+  }
+
+}

packages/grpc/test/index.test.ts

@@ -254,6 +254,104 @@ describe('/test/index.test.ts', function () {
     await closeApp(app);
   });
 
+  it('should support publish stream metadata gRPC server', async () => {
+    const app = await createServer('base-app-stream-meta');
+
+    const service = await createGRPCConsumer<math.MathClient>({
+      package: 'math',
+      protoPath: join(__dirname, 'fixtures/proto/math.proto'),
+      url: 'localhost:6571'
+    });
+
+    // 使用发送消息的写法
+    let result1 = await service.add().sendMessage({
+      num: 2,
+    });
+
+    expect(result1.num).toEqual(4);
+
+    // 服务端推送
+    let total = 0;
+    let result2 = await service.sumMany().sendMessage({
+      num: 1,
+    });
+
+    result2.forEach(data => {
+      total += data.num;
+    });
+
+    expect(total).toEqual(9);
+
+    // 客户端推送
+    const data = await service.addMany()
+      .sendMessage({num: 1})
+      .sendMessage({num: 2})
+      .sendMessage({num: 3})
+      .end();
+
+    expect(data.num).toEqual(6);
+
+    // 双向流
+    const result3= await new Promise<number>((resolve, reject) => {
+      const duplexCall = service.addMore().getCall();
+      total = 0;
+      let idx = 0;
+
+      duplexCall.on('data', (data: math.Num) => {
+        total += data.num;
+        idx++;
+        if (idx === 2) {
+          duplexCall.end();
+          resolve(total);
+        }
+      });
+
+      duplexCall.write({
+        num: 3,
+      });
+
+      duplexCall.write({
+        num: 6,
+      });
+    });
+
+    expect(result3).toEqual(29);
+
+
+    // 保证顺序的双向流
+    const t = service.addMore({
+      messageKey: 'id'
+    });
+
+    const result4 = await new Promise<number>((resolve, reject) => {
+      total = 0;
+      t.sendMessage({
+        num: 2
+      })
+        .then(res => {
+          expect(res.num).toEqual(12);
+          total += res.num;
+        })
+        .catch(err => console.error(err))
+      ;
+      t.sendMessage({
+        num: 5
+      })
+        .then(res => {
+          expect(res.num).toEqual(15);
+          total += res.num;
+          resolve(total);
+        })
+        .catch(err => console.error(err))
+      ;
+      t.end();
+    });
+
+    expect(result4).toEqual(27);
+
+    await closeApp(app);
+  });
+
   it('should test multi-package service', async () => {
     const app = await createServer('base-app-multiple-package');