How to use the wonka.filter function in wonka

? !!params.isClient
        : !client.suspense;

    const sharedOps$ = share(ops$);

    let forwardedOps$ = pipe(
      sharedOps$,
      filter(op => !isCached(op)),
      forward
    );

    // NOTE: Since below we might delete the cached entry after accessing
    // it once, cachedOps$ needs to be merged after forwardedOps$
    let cachedOps$ = pipe(
      sharedOps$,
      filter(op => isCached(op)),
      map(op => {
        const serialized = data[op.key];
        return deserializeResult(op, serialized);
      })
    );

    if (!isClient) {
      // On the server we cache results in the cache as they're resolved
      forwardedOps$ = pipe(
        forwardedOps$,
        tap((result: OperationResult) => {
          const { operation } = result;
          if (!shouldSkip(operation)) {
            const serialized = serializeResult(result);
            data[operation.key] = serialized;
          }

this.results$,
      filter(res => res.operation.key === key)
    );

    if (operationName === 'mutation') {
      // A mutation is always limited to just a single result and is never shared
      return pipe(
        operationResults$,
        onStart(() => this.dispatchOperation(operation)),
        take(1)
      );
    }

    const teardown$ = pipe(
      this.operations$,
      filter(op => op.operationName === 'teardown' && op.key === key)
    );

    const result$ = pipe(
      operationResults$,
      takeUntil(teardown$),
      onStart(() => this.onOperationStart(operation)),
      onEnd(() => this.onOperationEnd(operation))
    );

    return operation.context.suspense !== false &&
      this.suspense &&
      operationName === 'query'
      ? toSuspenseSource(result$)
      : result$;
  }

executeRequestOperation(operation: Operation): Source {
    const { key, operationName } = operation;
    const operationResults$ = pipe(
      this.results$,
      filter(res => res.operation.key === key)
    );

    if (operationName === 'mutation') {
      // A mutation is always limited to just a single result and is never shared
      return pipe(
        operationResults$,
        onStart(() => this.dispatchOperation(operation)),
        take(1)
      );
    }

    const teardown$ = pipe(
      this.operations$,
      filter(op => op.operationName === 'teardown' && op.key === key)
    );

o.variables,
            o.operationName,
            fieldResolver,
            typeResolver
          );
          return makeResult(o, r);
        } catch (err) {
          return makeErrorResult(o, err);
        }
      }),
      mergeMap(p => fromPromise(p))
    );

    const forwardedOps$ = pipe(
      sharedOps$,
      filter(o => !targetOperationTypes.includes(o.operationName)),
      forward
    );

    return merge([executedOps$, forwardedOps$]);
  };
};

const { key } = operation;
        const teardown$ = pipe(
          sharedOps$,
          filter(op => op.operationName === 'teardown' && op.key === key)
        );

        return pipe(
          createFetchSource(operation),
          takeUntil(teardown$)
        );
      })
    );

    const forward$ = pipe(
      sharedOps$,
      filter(op => !isOperationFetchable(op)),
      forward
    );

    return merge([fetchResults$, forward$]);
  };
};

return ops$ => {
    const sharedOps$ = share(ops$);
    const subscriptionResults$ = pipe(
      sharedOps$,
      filter(isSubscriptionOperation),
      mergeMap(operation => {
        const { key } = operation;
        const teardown$ = pipe(
          sharedOps$,
          filter(op => op.operationName === 'teardown' && op.key === key)
        );

        return pipe(
          createSubscriptionSource(operation),
          takeUntil(teardown$)
        );
      })
    );

    const forward$ = pipe(
      sharedOps$,

const makeQueryResults$ = (
  client: Client,
  queryProp$: Source
): Source => {
  const noopQuery = empty as Source;
  let lastResults$: void | Source;

  return pipe(
    queryProp$,
    map(query => {
      return query === undefined ? noopQuery : client.executeQuery(query);
    }),
    filter(x => {
      const isDistinct = x !== lastResults$;
      if (isDistinct) {
        lastResults$ = x;
      }
      return isDistinct;
    }),
    switchAll
  );
};

};

        if (operation.context.requestPolicy === 'cache-and-network') {
          result.fetching = true;
          reexecuteOperation(client, operation);
        }

        return result;
      })
    );

    const forwardedOps$ = pipe(
      merge([
        pipe(
          sharedOps$,
          filter(op => !shouldSkip(op) && !isOperationCached(op)),
          map(mapTypeNames)
        ),
        pipe(
          sharedOps$,
          filter(op => shouldSkip(op))
        ),
      ]),
      map(op => addMetadata(op, { cacheOutcome: 'miss' })),
      forward,
      tap(response => {
        if (
          response.operation &&
          response.operation.operationName === 'mutation'
        ) {
          handleAfterMutation(response);
        } else if (

const { key } = operation;
        const teardown$ = pipe(
          sharedOps$,
          filter(op => op.operationName === 'teardown' && op.key === key)
        );

        return pipe(
          createSubscriptionSource(operation),
          takeUntil(teardown$)
        );
      })
    );

    const forward$ = pipe(
      sharedOps$,
      filter(op => !isSubscriptionOperation(op)),
      forward
    );

    return merge([subscriptionResults$, forward$]);
  };
};

return ops$ => {
    const forward$ = pipe(
      ops$,
      filter(filterIncomingOperation)
    );
    return pipe(
      forward(forward$),
      tap(afterOperationResult)
    );
  };
};