How to use the wonka.pipe function in wonka

const ssr: SSRExchange = ({ client, forward }) => ops$ => {
    // params.isClient tells us whether we're on the client-side
    // By default we assume that we're on the client if suspense-mode is disabled
    const isClient =
      params && typeof params.isClient === 'boolean'
        ? !!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);
      })
    );

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)

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)
    );

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

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)
    );

    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$;

const isClient =
      params && typeof params.isClient === 'boolean'
        ? !!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);

const handleMessage = (client: Client) => (message: any) => {
  if (message.type === "request") {
    const isMutation = /(^|\W)+mutation\W/.test(message.query);
    const execFn = isMutation ? client.executeMutation : client.executeQuery;

    pipe(
      execFn(createRequest(message.query), {
        meta: { source: "Devtools" }
      }),
      toPromise
    );
  }
};

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

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

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

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

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