fix(microservices): fix grpc stream method return type

Ensure GrpcStreamMethod returns Observable directly instead of wrapping it in a Promise when called locally.

Closes #15953

packages/microservices/decorators/message-pattern.decorator.ts

@@ -152,16 +152,18 @@ export function GrpcStreamMethod(
 
     // Override original method to call the "drainBuffer" method on the first parameter
     // This is required to avoid premature message emission
-    descriptor.value = async function (
-      this: any,
-      observable: any,
-      ...args: any[]
-    ) {
-      const result = await Promise.resolve(
-        originalMethod.apply(this, [observable, ...args]),
-      );
+    descriptor.value = function (this: any, observable: any, ...args: any[]) {
+      const result = originalMethod.apply(this, [observable, ...args]);
+      const isPromise = result && typeof result.then === 'function';
+      if (isPromise) {
+        return result.then((data: any) => {
+          if (observable && observable.drainBuffer) {
+            observable.drainBuffer();
+          }
+          return data;
+        });
+      }
 
-      // Drain buffer if "drainBuffer" method is available
       if (observable && observable.drainBuffer) {
         observable.drainBuffer();
       }

packages/microservices/test/decorators/message-pattern.decorator.spec.ts

@@ -247,6 +247,18 @@ describe('@GrpcStreamMethod', () => {
       streaming: GrpcMethodStreamingType.RX_STREAMING,
     });
   });
+
+  it('should return Observable directly (not wrapped in Promise) when called method directly', () => {
+    class TestService {
+      @GrpcStreamMethod()
+      test(data$: any) {
+        return data$;
+      }
+    }
+    const service = new TestService();
+    const result = service.test({});
+    expect(result).to.not.have.property('then');
+  });
 });
 
 describe('@GrpcStreamCall', () => {