Angular SchemaLink

angular-schema-link.ts

import { Observable, Observer, Subscription } from 'rxjs'

import { Operation, FetchResult, Observable as ApolloLinkObservable } from 'apollo-link'

import { SchemaLink } from 'apollo-link-schema'

// adapted copy of https://github.com/angular/angular/blob/master/packages/platform-server/src/http.ts
declare var Zone: any
type Task = any

abstract class ZoneMacroTaskWrapper<S, R> {
  wrap(request: S): Observable<R> {
    return new Observable((observer: Observer<R>) => {
      let task: Task = null
      let scheduled = false
      let sub: Subscription | null = null
      let savedResult: any = null
      let savedError: any = null

      const scheduleTask = (_task: Task) => {
        task = _task
        scheduled = true

        const delegate = this.delegate(request)
        sub = delegate.subscribe(
          res => (savedResult = res),
          err => {
            if (!scheduled) {
              throw new Error(`A GraphQL observable was completed twice. This shouldn't happen, please file a bug.`)
            }
            savedError = err
            scheduled = false
            task.invoke()
          },
          () => {
            if (!scheduled) {
              throw new Error(`An GraphQL observable was completed twice. This shouldn't happen, please file a bug.`)
            }
            scheduled = false
            task.invoke()
          }
        )
      }

      const cancelTask = (_task: Task) => {
        if (!scheduled) {
          return
        }
        scheduled = false
        if (sub) {
          sub.unsubscribe()
          sub = null
        }
      }

      const onComplete = () => {
        if (savedError !== null) {
          observer.error(savedError)
        } else {
          observer.next(savedResult)
          observer.complete()
        }
      }

      // MockBackend for Http is synchronous, which means that if scheduleTask is by
      // scheduleMacroTask, the request will hit MockBackend and the response will be
      // sent, causing task.invoke() to be called.
      const _t = Zone.current.scheduleMacroTask(
        'ZoneMacroTaskWrapper.subscribe',
        onComplete,
        {},
        () => null,
        cancelTask
      )
      scheduleTask(_t)

      return () => {
        if (scheduled && task) {
          task.zone.cancelTask(task)
          scheduled = false
        }
        if (sub) {
          sub.unsubscribe()
          sub = null
        }
      }
    })
  }

  protected abstract delegate(request: S): Observable<R>
}

class SchemaLinkRequestZoneMacroTask extends ZoneMacroTaskWrapper<Operation, FetchResult> {
  constructor(private schemaLink: AngularSchemaLink) {
    super()
  }

  request(op: Operation): Observable<FetchResult> {
    return this.wrap(op)
  }

  protected delegate(op: Operation): Observable<FetchResult> {
    return new Observable<FetchResult>(obs => {
      const sub = this.schemaLink.requestToDelegate(op).subscribe(
        v => obs.next(v),
        e => obs.error(e),
        () => obs.complete()
      )
      return () => sub.unsubscribe()
    })
  }
}

export class AngularSchemaLink extends SchemaLink {
  private zoneMacroTask: SchemaLinkRequestZoneMacroTask = new SchemaLinkRequestZoneMacroTask(this)

  constructor(options: SchemaLink.Options) {
    super(options)
  }

  requestToDelegate(operation: Operation): ApolloLinkObservable<FetchResult> {
    const task = super.request(operation)
    // for SchemaLink it should not be the case
    // but ApolloLink defines a possible null return
    // just to make typescript happy
    if (task == null) {
      return new ApolloLinkObservable<FetchResult>(sub => {
        sub.error(new Error('SchemaLink request returns null observable'))
      })
    }
    return task
  }

  request(operation: Operation): ApolloLinkObservable<FetchResult> {
    return new ApolloLinkObservable<FetchResult>(obs => {
      const sub = this.zoneMacroTask.request(operation).subscribe(
        v => obs.next(v),
        e => obs.error(e),
        () => obs.complete()
      )
      return () => sub.unsubscribe()
    })
  }
}