Fix dedup ignoring torn down query operations (#281)

When a query was torn down, the dedup exchange wasn't
removing its key from the inFlight keys.

This was causing us to ignore torn down queries forever.

src/exchanges/dedup.test.ts

@@ -75,6 +75,19 @@ it('forwards duplicate query operations as usual after they respond', async () =
   expect(forwardedOperations.length).toBe(2);
 });
 
+it('forwards duplicate query operations after one was torn down', async () => {
+  shouldRespond = false; // We filter out our mock responses
+  const [ops$, next, complete] = input;
+  const exchange = dedupExchange(exchangeArgs)(ops$);
+
+  publish(exchange);
+  next(queryOperation);
+  next({ ...queryOperation, operationName: 'teardown' });
+  next(queryOperation);
+  complete();
+  expect(forwardedOperations.length).toBe(3);
+});
+
 it('always forwards mutation operations without deduplicating them', async () => {
   shouldRespond = false; // We filter out our mock responses
   const [ops$, next, complete] = input;

src/exchanges/dedup.ts

@@ -1,32 +1,36 @@
 import { filter, pipe, tap } from 'wonka';
-import { Exchange } from '../types';
+import { Exchange, Operation, OperationResult } from '../types';
 
 /** A default exchange for debouncing GraphQL requests. */
 export const dedupExchange: Exchange = ({ forward }) => {
-  const inFlight = new Set<number>();
+  const inFlightKeys = new Set<number>();
 
-  return ops$ =>
-    pipe(
-      forward(
-        pipe(
-          ops$,
-          filter(({ operationName, key }) => {
-            if (operationName !== 'query') {
-              return true;
-            }
+  const filterIncomingOperation = (operation: Operation) => {
+    const { key, operationName } = operation;
+    if (operationName === 'teardown') {
+      inFlightKeys.delete(key);
+      return true;
+    } else if (operationName !== 'query') {
+      return true;
+    }
 
-            const hasInFlightOp = inFlight.has(key);
+    const isInFlight = inFlightKeys.has(key);
+    inFlightKeys.add(key);
+    return !isInFlight;
+  };
 
-            if (!hasInFlightOp) {
-              inFlight.add(key);
-            }
+  const afterOperationResult = ({ operation }: OperationResult) => {
+    inFlightKeys.delete(operation.key);
+  };
 
-            return !hasInFlightOp;
-          })
-        )
-      ),
-      tap(res => {
-        inFlight.delete(res.operation.key);
-      })
+  return ops$ => {
+    const forward$ = pipe(
+      ops$,
+      filter(filterIncomingOperation)
     );
+    return pipe(
+      forward(forward$),
+      tap(afterOperationResult)
+    );
+  };
 };