Remove wait-for-observables in favor of ObservableStream (apollographql#12317)

Author: jerelmiller

package-lock.json

@@ -198,7 +198,6 @@
     "tsx": "4.19.2",
     "typedoc": "0.25.0",
     "typescript": "5.7.3",
-    "wait-for-observables": "1.0.3",
     "web-streams-polyfill": "4.0.0",
     "whatwg-fetch": "3.6.20"
   },

package.json

@@ -1,11 +1,11 @@
 import gql from "graphql-tag";
-import waitFor from "wait-for-observables";
 
 import { ApolloLink } from "../../core/ApolloLink";
 import { execute } from "../../core/execute";
 import { Observable } from "../../../utilities/observables/Observable";
 import { fromError } from "../../utils/fromError";
 import { RetryLink } from "../retryLink";
+import { ObservableStream } from "../../../testing/internal";
 
 const query = gql`
   {
@@ -23,9 +23,10 @@ describe("RetryLink", () => {
     const retry = new RetryLink({ delay: { initial: 1 }, attempts: { max } });
     const stub = jest.fn(() => fromError(standardError)) as any;
     const link = ApolloLink.from([retry, stub]);
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitError(standardError, { timeout: 1000 });
 
-    const [{ error }] = (await waitFor(execute(link, { query }))) as any;
-    expect(error).toEqual(standardError);
     expect(stub).toHaveBeenCalledTimes(max);
   });
 
@@ -34,9 +35,11 @@ describe("RetryLink", () => {
     const data = { data: { hello: "world" } };
     const stub = jest.fn(() => Observable.of(data));
     const link = ApolloLink.from([retry, stub]);
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitValue(data);
+    await expect(stream).toComplete();
 
-    const [{ values }] = (await waitFor(execute(link, { query }))) as any;
-    expect(values).toEqual([data]);
     expect(stub).toHaveBeenCalledTimes(1);
   });
 
@@ -50,9 +53,11 @@ describe("RetryLink", () => {
     stub.mockReturnValueOnce(fromError(standardError));
     stub.mockReturnValueOnce(Observable.of(data));
     const link = ApolloLink.from([retry, stub]);
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitValue(data);
+    await expect(stream).toComplete();
 
-    const [{ values }] = (await waitFor(execute(link, { query }))) as any;
-    expect(values).toEqual([data]);
     expect(stub).toHaveBeenCalledTimes(2);
   });
 
@@ -129,13 +134,14 @@ describe("RetryLink", () => {
     });
     const stub = jest.fn(() => fromError(standardError)) as any;
     const link = ApolloLink.from([retry, stub]);
+    const stream1 = new ObservableStream(execute(link, { query }));
+    const stream2 = new ObservableStream(execute(link, { query }));
+
+    await Promise.all([
+      expect(stream1).toEmitError(standardError),
+      expect(stream2).toEmitError(standardError),
+    ]);
 
-    const [result1, result2] = (await waitFor(
-      execute(link, { query }),
-      execute(link, { query })
-    )) as any;
-    expect(result1.error).toEqual(standardError);
-    expect(result2.error).toEqual(standardError);
     expect(stub).toHaveBeenCalledTimes(10);
   });
 
@@ -144,9 +150,10 @@ describe("RetryLink", () => {
     const retry = new RetryLink({ delay: delayStub, attempts: { max: 3 } });
     const linkStub = jest.fn(() => fromError(standardError)) as any;
     const link = ApolloLink.from([retry, linkStub]);
-    const [{ error }] = (await waitFor(execute(link, { query }))) as any;
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitError(standardError);
 
-    expect(error).toEqual(standardError);
     const operation = (delayStub.mock.calls[0] as any)[1];
     expect(delayStub.mock.calls).toEqual([
       [1, operation, standardError],
@@ -166,9 +173,10 @@ describe("RetryLink", () => {
     });
     const linkStub = jest.fn(() => fromError(standardError)) as any;
     const link = ApolloLink.from([retry, linkStub]);
-    const [{ error }] = (await waitFor(execute(link, { query }))) as any;
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitError(standardError);
 
-    expect(error).toEqual(standardError);
     const operation = attemptStub.mock.calls[0][1];
     expect(attemptStub.mock.calls).toEqual([
       [1, operation, standardError],
@@ -191,9 +199,10 @@ describe("RetryLink", () => {
       () => new Observable((o) => o.error(standardError))
     ) as any;
     const link = ApolloLink.from([retry, linkStub]);
-    const [{ error }] = (await waitFor(execute(link, { query }))) as any;
+    const stream = new ObservableStream(execute(link, { query }));
+
+    await expect(stream).toEmitError(standardError);
 
-    expect(error).toEqual(standardError);
     const operation = attemptStub.mock.calls[0][1];
     expect(attemptStub.mock.calls).toEqual([
       [1, operation, standardError],