How to use the wonka.tap function in wonka

return ops$ =>
      pipe(
        ops$,
        // eslint-disable-next-line no-console
        tap(op => console.log('[Exchange debug]: Incoming operation: ', op)),
        forward,
        tap(result =>
          // eslint-disable-next-line no-console
          console.log('[Exchange debug]: Completed operation: ', result)
        )
      );
  }

return ops$ =>
      pipe(
        ops$,
        // eslint-disable-next-line no-console
        tap(op => console.log('[Exchange debug]: Incoming operation: ', op)),
        forward,
        tap(result =>
          // eslint-disable-next-line no-console
          console.log('[Exchange debug]: Completed operation: ', result)
        )
      );
  }

// 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;
          }
        })
      );
    } else {
      // On the client we delete results from the cache as they're resolved
      cachedOps$ = pipe(
        cachedOps$,
        tap((result: OperationResult) => {
          delete data[result.operation.key];
        })
      );
    }

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

return ops$ => {
    return pipe(
      ops$,
      map(addOperationContext),
      tap(handleOperation),
      forward,
      map(addOperationResponseContext),
      tap(handleOperation)
    );
  };
};

return ops$ => {
    return pipe(
      ops$,
      map(addOperationContext),
      tap(handleOperation),
      forward,
      map(addOperationResponseContext),
      tap(handleOperation)
    );
  };
};

return ops$ => {
    return pipe(
      ops$,
      tap(handleIncomingQuery),
      tap(handleIncomingTeardown),
      map(handleIncomingMutation),
      forward
    );
  };
};

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 (
          response.operation &&
          response.operation.operationName === 'query'
        ) {
          handleAfterQuery(response);
        }
      })
    );

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

return ops$ => {
    return pipe(
      ops$,
      tap(handleIncomingQuery),
      tap(handleIncomingTeardown),
      map(handleIncomingMutation),
      forward
    );
  };
};