fix(core): ensure the type `T` of `EventEmitter<T>` can be inferred (#40644)

The `AsyncPipe.transform<T>(emitter)` method must infer the `T`
type from the `emitter` parameter. Since we changed the `AsyncPipe`
to expect a `Subscribable<T>` rather than `Observable<T>` the
`EventEmitter.subscribe()` method needs to have a tighter signature.
Otherwise TypeScript struggles to infer the type and ends up making
it `unknown`.

Fixes #40637

PR Close #40644

goldens/public-api/core/core.d.ts

@@ -314,7 +314,8 @@ export declare class ErrorHandler {
 export declare interface EventEmitter<T> extends Subject<T> {
     new (isAsync?: boolean): EventEmitter<T>;
     emit(value?: T): void;
-    subscribe(generatorOrNext?: any, error?: any, complete?: any): Subscription;
+    subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void): Subscription;
+    subscribe(observerOrNext?: any, error?: any, complete?: any): Subscription;
 }
 
 export declare const EventEmitter: {

packages/common/test/pipes/async_pipe_spec.ts

@@ -129,6 +129,16 @@ import {SpyChangeDetectorRef} from '../spies';
       });
     });
 
+    describe('Subscribable', () => {
+      it('should infer the type from the subscribable', () => {
+        const ref = new SpyChangeDetectorRef() as any;
+        const pipe = new AsyncPipe(ref);
+        const emitter = new EventEmitter<{name: 'T'}>();
+        // The following line will fail to compile if the type cannot be inferred.
+        const name = pipe.transform(emitter)?.name;
+      });
+    });
+
     describe('Promise', () => {
       const message = {};
       let pipe: AsyncPipe;

packages/compiler-cli/src/ngtsc/testing/fake_core/index.ts

@@ -90,7 +90,10 @@ export const CUSTOM_ELEMENTS_SCHEMA: any = false;
 export const NO_ERRORS_SCHEMA: any = false;
 
 export class EventEmitter<T> {
-  subscribe(generatorOrNext?: any, error?: any, complete?: any): unknown {
+  subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void):
+      unknown;
+  subscribe(observerOrNext?: any, error?: any, complete?: any): unknown;
+  subscribe(observerOrNext?: any, error?: any, complete?: any): unknown {
     return null;
   }
 }

packages/compiler-cli/src/ngtsc/typecheck/test/test_utils.ts

@@ -97,7 +97,8 @@ export function angularCoreDts(): TestFile {
     }
 
     export declare class EventEmitter<T> {
-      subscribe(generatorOrNext?: any, error?: any, complete?: any): unknown;
+      subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void): unknown;
+      subscribe(observerOrNext?: any, error?: any, complete?: any): unknown;
     }
 
     export declare type NgIterable<T> = Array<T> | Iterable<T>;

packages/core/src/event_emitter.ts

@@ -81,15 +81,25 @@ export interface EventEmitter<T> extends Subject<T> {
    * @param value The value to emit.
    */
   emit(value?: T): void;
+
   /**
    * Registers handlers for events emitted by this instance.
-   * @param generatorOrNext When supplied, a custom handler for emitted events.
+   * @param next When supplied, a custom handler for emitted events.
-   * @param error When supplied, a custom handler for an error notification
+   * @param error When supplied, a custom handler for an error notification from this emitter.
-   * from this emitter.
+   * @param complete When supplied, a custom handler for a completion notification from this
-   * @param complete When supplied, a custom handler for a completion
+   *     emitter.
-   * notification from this emitter.
    */
-  subscribe(generatorOrNext?: any, error?: any, complete?: any): Subscription;
+  subscribe(next?: (value: T) => void, error?: (error: any) => void, complete?: () => void):
+      Subscription;
+  /**
+   * Registers handlers for events emitted by this instance.
+   * @param observerOrNext When supplied, a custom handler for emitted events, or an observer
+   *     object.
+   * @param error When supplied, a custom handler for an error notification from this emitter.
+   * @param complete When supplied, a custom handler for a completion notification from this
+   *     emitter.
+   */
+  subscribe(observerOrNext?: any, error?: any, complete?: any): Subscription;
 }
 
 class EventEmitter_ extends Subject<any> {
@@ -104,38 +114,38 @@ class EventEmitter_ extends Subject<any> {
     super.next(value);
   }
 
-  subscribe(generatorOrNext?: any, error?: any, complete?: any): Subscription {
+  subscribe(observerOrNext?: any, error?: any, complete?: any): Subscription {
     let schedulerFn: (t: any) => any;
     let errorFn = (err: any): any => null;
     let completeFn = (): any => null;
 
-    if (generatorOrNext && typeof generatorOrNext === 'object') {
+    if (observerOrNext && typeof observerOrNext === 'object') {
       schedulerFn = this.__isAsync ? (value: any) => {
-        setTimeout(() => generatorOrNext.next(value));
+        setTimeout(() => observerOrNext.next(value));
       } : (value: any) => {
-        generatorOrNext.next(value);
+        observerOrNext.next(value);
       };
 
-      if (generatorOrNext.error) {
+      if (observerOrNext.error) {
         errorFn = this.__isAsync ? (err) => {
-          setTimeout(() => generatorOrNext.error(err));
+          setTimeout(() => observerOrNext.error(err));
         } : (err) => {
-          generatorOrNext.error(err);
+          observerOrNext.error(err);
         };
       }
 
-      if (generatorOrNext.complete) {
+      if (observerOrNext.complete) {
         completeFn = this.__isAsync ? () => {
-          setTimeout(() => generatorOrNext.complete());
+          setTimeout(() => observerOrNext.complete());
         } : () => {
-          generatorOrNext.complete();
+          observerOrNext.complete();
         };
       }
     } else {
       schedulerFn = this.__isAsync ? (value: any) => {
-        setTimeout(() => generatorOrNext(value));
+        setTimeout(() => observerOrNext(value));
       } : (value: any) => {
-        generatorOrNext(value);
+        observerOrNext(value);
       };
 
       if (error) {
@@ -157,8 +167,8 @@ class EventEmitter_ extends Subject<any> {
 
     const sink = super.subscribe(schedulerFn, errorFn, completeFn);
 
-    if (generatorOrNext instanceof Subscription) {
+    if (observerOrNext instanceof Subscription) {
-      generatorOrNext.add(sink);
+      observerOrNext.add(sink);
     }
 
     return sink;