feat(GRPC): Stream decorator and stream handler pass-through for Controllers to support GRPC Duplex streams defined with Protobuf

[anton-alation]

PR Checklist

Please check if your PR fulfills the following requirements:

• The commit message follows our guidelines: https://github.com/nestjs/nest/blob/master/CONTRIBUTING.md
• Tests for the changes have been added (for bug fixes / features)
• Docs have been added / updated (for bug fixes / features)

PR Type

What kind of change does this PR introduce?

[ ] Bugfix
[X] Feature
[ ] Code style update (formatting, local variables)
[X] Refactoring (no functional changes, no api changes)
[ ] Build related changes
[ ] CI related changes
[ ] Other... Please describe:

What is the current behavior?

Issue Number: #1286
Issue Number: #1264
Previous PR: #1292

Currently when protobuf rpc is defined like this:

service TestService {
    rpc testMethod(stream TestMessage) returns (stream TestResponse)
}

Nest has no ability to pass a stream object down to the Controller handler, so there is no option to have full-duplex message interaction.

What is the new behavior?

Two new behaviors added, both are aiming same goals but serves different approach towards reaching the goal. Let's enumerate those:

RX Object approach

Developer receives RX object as the message and can return Promise or RX object as the stream

@Controller()
export class GrpcController {
  @GrpcStreamMethod('TestService')
    async testMethod(messages: Observable<TestMessage>): Promise<TestResponse> {
      // Form a resulting promise
      return new Promise<any>((resolve, reject) => {
        // Subscribe for a message to for a test answer
        messages.subscribe(msg => {
          // Resolve with iterating the value
          resolve({
            result: msg.data + 1
          });
        }, err => {
          reject(err);
        });
      });
    }
}

GRPC Call Object approach

Developer receives GRPC Stream wrapper and can operate it in the style of how streams are operated:

@Controller()
export class GrpcController {
    @GrpcStreamCall('TestService')
    async testMethod(stream: ClientWritableStream<TestResponse>) {
        stream.on('data', (msg: TestMessage) => {
            stream.write({result: msg.data + 1});
        }
    }
}

Does this PR introduce a breaking change?

[ ] Yes
[X] No

Other information

[coveralls]

Pull Request Test Coverage Report for Build 1543

• 15 of 19 (78.95%) changed or added relevant lines in 2 files are covered.
• No unchanged relevant lines lost coverage.
• Overall coverage decreased (-0.1%) to 93.705%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
packages/microservices/server/server-grpc.ts	7	11	63.64%

Totals
Change from base Build 1542:	-0.1%
Covered Lines:	2889
Relevant Lines:	3016

---

💛 - Coveralls

[coveralls]

Pull Request Test Coverage Report for Build 1609

• 23 of 51 (45.1%) changed or added relevant lines in 2 files are covered.
• 5 unchanged lines in 2 files lost coverage.
• Overall coverage decreased (-0.2%) to 93.618%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
packages/microservices/decorators/pattern.decorator.ts	13	16	81.25%
packages/microservices/server/server-grpc.ts	10	35	28.57%

Files with Coverage Reduction	New Missed Lines	%
packages/core/router/routes-resolver.ts	1	92.16%
packages/microservices/client/client-rmq.ts	4	93.24%

Totals
Change from base Build 1542:	-0.2%
Covered Lines:	2654
Relevant Lines:	2798

---

💛 - Coveralls

[mkaufmaner]

@anton-alation First of all, GREAT WORK! I noticed that you decided to go with @GrpcStream for the decorator. However, I feel as if nestjs should be wrapping the stream with an Observable to maintain continuity. That being the case, maybe rename @GrpcStream to @GrpcStreamPassthrough?

[anton-alation]

@mkaufmaner @kamilmysliwiec Thank you. So here should be some consensus on what we trying to achieve because converting the stream to Observable would be just excessive wrapping of an already simple interface with just .on() and .cancel() methods.

However if let's say I would add something like:

return async (call) => {
      const req = new Subject<any>();
      // Pass data to request
      call.on('data', (m: any) => req.next(m));
      // React on error
      call.on('error', (e: any) => {
        // Check if error means that stream ended on other end
        if (String(e).toLowerCase().indexOf('cancelled') > -1) {
          call.end();
          req.complete();
          return;
        }
        // If another error then just pass it along
        req.error(e);
      });
      // Pass request to handler
      const handler = methodHandler(req.asObservable());
      // Receive back observable
      const observable = this.transformToObservable(await handler);
      // Write until cancelled event happened
      await observable.pipe(takeUntil(fromEvent(call, CANCEL_EVENT)))
        .forEach(msg => call.write(msg));
      call.end();
};

Would if be more satisfactory behavior? Only significant benefit of that I see in having predefined Observable interface rather than maybe least known GRPC interface Duplex stream:

@GrpcStream('TestMethod')
  async testStream(stream: Observable<TestMessage>) {
    // Implementation
  }

Please thoughts and arguments toward one or other approach.

For my current projects, it's cheaper to have PassThrough just because this handler passed down to Abstracted executor layer which do all of the Job.

But if we need to be a bit fancier we can work on as PassThrough, as well on Rx Observable passed down.

[mkaufmaner]

@anton-alation I think we should have both the passthrough and the observable. The observable does simplify implementation with GRPC while also providing a rich feature set.

return async (call, methodHandler) => {
  const req = new Subject<any>();
  // Pass data to request
  call.on('data', (m: any) => req.next(m));
  // React on error
  call.on('error', (e: any) => {
    // Check if error means that stream ended on other end
    if (String(e).toLowerCase().indexOf('cancelled') > -1) {
      call.end();
      return;
    }
    // If another error then just pass it along
    req.error(e);
  });

  call.on('end', () => req.complete());

  // Pass request to handler
  const handler = methodHandler(req.asObservable());
  // Receive back observable
  const res = this.transformToObservable(await handler);
  // Write until cancelled event happened
  await res.pipe(takeUntil(fromEvent(call, CANCEL_EVENT))).forEach(m => call.write(m));

  call.end();
};

[anton-alation]

@mkaufmaner Let me add additional annotation and write tests to check that Rx behavior will work fine wrapping stream things. Later this week I will provide an update.

[anton-alation]

@mkaufmaner @kamilmysliwiec Solution is in CI smoke now, locally it was fine. Will be editing PR documentation on naming. Feel free to review :)

[anton-alation]

@mkaufmaner @kamilmysliwiec any news or updates?

[kamilmysliwiec]

Hey @anton-alation,
I'm quite busy due to the v6 release. I'll review shortly. Thanks :)

[anton-alation]

Hey @anton-alation,
I'm quite busy due to the v6 release. I'll review shortly. Thanks :)

Thanks :) wasn't sure if patterns would be reviewed in close future, while the feature is very important. I understand the load :)

[PhilipMantrov]

Hey @kamilmysliwiec, this feature is really important for me so if you need some help i can help you asap.

[anton-alation]

@PhilipMantrov with a few hacks (runtime-patches) you can go around current Nest behavior, but because we added this PR probably it's also very reasonable to just wait for the feature to merge.

[kamilmysliwiec]

Hey @anton-alation,
I pulled your changes, added unit tests, resolved conflicts and I'm ready to merge this feature :) Would like to create a PR to the docs with those new decorators (and briefly explain their behavior)? Simply edit this markdown file: https://github.com/nestjs/docs.nestjs.com/blob/master/content/microservices/grpc.md

[anton-alation]

Hey @anton-alation,
I pulled your changes, added unit tests, resolved conflicts and I'm ready to merge this feature :) Would like to create a PR to the docs with those new decorators (and briefly explain their behavior)? Simply edit this markdown file: https://github.com/nestjs/docs.nestjs.com/blob/master/content/microservices/grpc.md

Thanks @kamilmysliwiec , please excuse me on losing track on master branch changes :)

Sure, let me do this updates to markdown tomorrow :)

[kamilmysliwiec]

Thanks @kamilmysliwiec , please excuse me on losing track on master branch changes :)

No worries, it's my fault, I should look at this PR earlier!

Sure, let me do this updates to markdown tomorrow :)

Awesome :)

[anton-alation]

Thanks @kamilmysliwiec , please excuse me on losing track on master branch changes :)

No worries, it's my fault, I should look at this PR earlier!

Sure, let me do this updates to markdown tomorrow :)

Awesome :)

@kamilmysliwiec Started with the doc here: commit for Docs, need to fulfill few outstanding topics which we made through the winter:

• Support multi-depth namespaces
• Support streaming
  Both joined under the section Advanced GRPC

Need a few more days. Any thoughts appreciated :)

[kamilmysliwiec]

Amazing @anton-alation, looking forward!

[kamilmysliwiec]

Did you have some time to look at the docs @anton-alation? Specifically at streaming part :) (required to merge this PR) Thanks!

[anton-alation]

Did you have some time to look at the docs @anton-alation? Specifically at streaming part :) (required to merge this PR) Thanks!

@kamilmysliwiec Yeah, I will be finalizing today and tomorrow. Last few weeks have collided with the GA release cycle of our product, so I was pretty much into making all things smooth and silky, which can be a bit exhaustive sometimes :)

[anton-alation]

So at the moment, I am writing tests to check that documentation I writing (almost wrote) would have the correct behavior within the nest, hope to finish tomorrow or Sunday.

[anton-alation]

@kamilmysliwiec created docs PR: nestjs/docs.nestjs.com#378 probably need for review.

As well I created a branch with actual tests related to those docs: #2075 targeting branch with the master updates.

[anton-alation]

@kamilmysliwiec Hi Kamil, can we expect this to be released soon? :)

[anton-alation]

@kamilmysliwiec any news on that one?

[mkaufmaner]

@anton-alation There seems to be some conflicts with the PR. Can you resolve?

[anton-alation]

@anton-alation There seems to be some conflicts with the PR. Can you resolve?

Hi Michael, I believe Kamil dedicated his efforts to resolve conflicts here:
https://github.com/nestjs/nest/tree/anton-alation-grpc-stream-duplex-decorator

Quoting:

Hey @anton-alation,
I pulled your changes, added unit tests, resolved conflicts and I'm ready to merge this feature :) Would like to create a PR to the docs with those new decorators (and briefly explain their behavior)? Simply edit this markdown file: https://github.com/nestjs/docs.nestjs.com/blob/master/content/microservices/grpc.md

As well as of discussion above there were updates to tests and docs.

Do you want me to synchronize exactly this PR branch to resolve conflicts?

[kamilmysliwiec]

@anton-alation I finished that part, no worries

[anton-alation]

@anton-alation I finished that part, no worries

Anything I can do for help with this feature or tests or docs?

[kamilmysliwiec]

This PR will be a part of the next 6.3.0 release :)

[anton-alation]

This PR will be a part of the next 6.3.0 release :)

Amazing! :) Thank you. If you need me to change something on the docs update, please tell :)

[kmturley]

I've updated my example using this new Streaming feature here:
https://github.com/kmturley/angular-nest-grpc

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.