return Object.keys(apps).reduce((states, key) => {
   const app = apps[key];
   const nextState = app(subject, initialState[key]);
   if (nextState.view) {
     if (!isObservable(nextState.view)) {
       throw new Error(`Application view of ${key} must be Observable.`);
     }
     nextState.view = nextState.view.pipe(startWith(initialState[key]));
     if (!states.view) {
       states.view = {};
     }
     if (!isDefined(states.view[key])) {
       states.view[key] = nextState.view;
     } else {
       states.view[key] = merge(states.view[key], nextState.view);
     }
   }
   for (const key in nextState) {
     if (key !== 'view') {
       if (!nextState[key] || !(nextState[key] instanceof Observable)) {
         throw new Error(`Property ${key} must be Observable.`);
       }
       if (states[key]) {
         states[key] = merge(states[key], nextState[key]);
       } else {
         states[key] = nextState[key];
       }
     }
   }
   return states;
 }, states);

  it('should NOT return true for any old subscribable', () => {
    const o: any = {
      subscribe() { /* noop */ },
    };

    expect(isObservable(o)).to.be.false;
  });

        directive.cleanup.add(component.subscribe(comp => {
            doCleanup();
            internal = new Subscription();

            // isolated nodestate and scope
            const scope = new Scope(comp.viewmodel);

            // wire custom events
            if (comp.viewmodel.emitter !== undefined && isObservable(comp.viewmodel.emitter)) {
                internal.add(comp.viewmodel.emitter.subscribe(evt => host.dispatchEvent(evt)));
            }
            // apply custom component value
            if (comp.viewmodel.value !== undefined && isObservable(comp.viewmodel.value)) {
                internal.add(comp.viewmodel.value.subscribe(val => {
                    host["value"] = val;
                    const evt = this.engine.createEvent("change");
                    host.dispatchEvent(evt);
                }));
            }
            // auto-dispose view-model
            if (comp.viewmodel.cleanup !== undefined) {
                internal.add(comp.viewmodel.cleanup);
            }

            this.applyTemplate(element, scope, comp, children);
            // created lifecycle hook
            if (comp.created !== undefined) {
                comp.created(element, scope);
            }

            internal.add(() => {
                if (comp.destroy !== undefined) {
                    comp.destroy(element, scope);
                }
                this.domManager.cleanDescendants(element);
                while (element.firstChild) {
                    element.removeChild(element.firstChild);
                }
            });
        }));

  static handle(process: any, {request, response, next, hasNextFunction}: any) {
    const done = (error: any, result?: any) => {
      if (error) {
        return next(error);
      }

      if (!hasNextFunction) {
        if (!next.isCalled && result !== undefined) {
          request.ctx.data = result;
        }
        next();
      }
    };

    if (process) {
      if (process === response) {
        return;
      }

      if (isObservable(process)) {
        // TODO basic support
        process = process.toPromise();
      }

      if (isStream(process)) {
        return done(null, process);
      }

      if (isFunction(process)) {
        // when process return a middleware
        return process(request, response, next);
      }

      if (isPromise(process)) {
        return process
          .then((result: any) =>
            this.handle(result, {
              request,
              response,
              next,
              hasNextFunction
            })
          )
          .catch((error: any) => done(error));
      }
    }

    if (!hasNextFunction) {
      // no next function and empty response
      done(null, process);
    }
  }

    it("should initialize correctly", () => {

        expect(model.disabled).toBe(false);
        expect(model.errorMessages).toBeNull();
        expect(model.hasErrorMessages).toBe(false);
        expect(model.hidden).toBe(false);
        expect(model.id).toEqual(config.id);
        expect(model.label).toBeNull();
        expect(model.legend).toBeNull();
        expect(isObservable(model.options$)).toBe(true);
        expect(model.type).toEqual(DYNAMIC_FORM_CONTROL_TYPE_RADIO_GROUP);
        expect(model.value).toBeNull();
    });

    function handleRequest(requestMessage: RequestMessage): void {
        if (isUnsubscribed()) {
            // we return here silently since we fired an event when the
            // connection got unsubscribed.
            return
        }

        const startTime = Date.now()

        function reply(resultOrError: any | ResponseError<any>, complete: boolean): void {
            const message: ResponseMessage = {
                jsonrpc: version,
                id: requestMessage.id,
                complete,
            }
            if (resultOrError instanceof ResponseError) {
                message.error = (resultOrError as ResponseError<any>).toJSON()
            } else {
                message.result = resultOrError === undefined ? null : resultOrError
            }
            tracer.responseSent(message, requestMessage, startTime)
            transports.writer.write(message)
        }
        function replyError(error: ResponseError<any>): void {
            const message: ResponseMessage = {
                jsonrpc: version,
                id: requestMessage.id,
                error: error.toJSON(),
                complete: true,
            }
            tracer.responseSent(message, requestMessage, startTime)
            transports.writer.write(message)
        }
        function replySuccess(result: any): void {
            // The JSON RPC defines that a response must either have a result or an error
            // So we can't treat undefined as a valid response result.
            if (result === undefined) {
                result = null
            }
            const message: ResponseMessage = {
                jsonrpc: version,
                id: requestMessage.id,
                result,
                complete: true,
            }
            tracer.responseSent(message, requestMessage, startTime)
            transports.writer.write(message)
        }
        function replyComplete(): void {
            const message: ResponseMessage = {
                jsonrpc: version,
                id: requestMessage.id,
                complete: true,
            }
            tracer.responseSent(message, requestMessage, startTime)
            transports.writer.write(message)
        }

        tracer.requestReceived(requestMessage)

        const element = requestHandlers[requestMessage.method]
        const requestHandler: GenericRequestHandler<any, any> | undefined = element && element.handler
        if (requestHandler || starRequestHandler) {
            const abortController = new AbortController()
            const signalKey = String(requestMessage.id)
            requestAbortControllers[signalKey] = abortController
            try {
                const params = requestMessage.params !== undefined ? requestMessage.params : null
                const handlerResult = requestHandler
                    ? requestHandler(params, abortController.signal)
                    : starRequestHandler!(requestMessage.method, params, abortController.signal)

                if (!handlerResult) {
                    delete requestAbortControllers[signalKey]
                    replySuccess(handlerResult)
                } else if (isPromise(handlerResult) || isObservable(handlerResult)) {
                    const onComplete = () => {
                        delete requestAbortControllers[signalKey]
                    }
                    from(handlerResult).subscribe(
                        value => reply(value, false),
                        error => {
                            onComplete()
                            if (error instanceof ResponseError) {
                                replyError(error as ResponseError<any>)
                            } else if (error && typeof error.message === 'string') {
                                replyError(
                                    new ResponseError<void>(ErrorCodes.InternalError, error.message, {
                                        stack: error.stack,
                                        ...error,
                                    })
                                )
                            } else {
                                replyError(
                                    new ResponseError<void>(
                                        ErrorCodes.InternalError,
                                        `Request ${
                                            requestMessage.method
                                        } failed unexpectedly without providing any details.`
                                    )
//.........这里部分代码省略.........

    return new Observable<JobOutboundMessage<O>>(subject => {
      // Handle input.
      inboundBus.subscribe(message => {
        switch (message.kind) {
          case JobInboundMessageKind.Ping:
            subject.next({ kind: JobOutboundMessageKind.Pong, description, id: message.id });
            break;

          case JobInboundMessageKind.Stop:
            // There's no way to cancel a promise or a synchronous function, but we do cancel
            // observables where possible.
            if (subscription) {
              subscription.unsubscribe();
            }
            subject.next({ kind: JobOutboundMessageKind.End, description });
            subject.complete();
            // Close all channels.
            channels.forEach(x => x.complete());
            break;

          case JobInboundMessageKind.Input:
            inputChannel.next(message.value);
            break;
        }
      });

      // Configure a logger to pass in as additional context.
      const logger = new Logger('job');
      logger.subscribe(entry => {
        subject.next({
          kind: JobOutboundMessageKind.Log,
          description,
          entry,
        });
      });

      // Execute the function with the additional context.
      subject.next({ kind: JobOutboundMessageKind.Start, description });

      const channels = new Map<string, Subject<JsonValue>>();

      const newContext = {
        ...context,
        input: inputChannel.asObservable(),
        logger,
        createChannel(name: string) {
          if (channels.has(name)) {
            throw new ChannelAlreadyExistException(name);
          }
          const channelSubject = new Subject<JsonValue>();
          channelSubject.subscribe(
            message => {
              subject.next({
                kind: JobOutboundMessageKind.ChannelMessage, description, name, message,
              });
            },
            error => {
              subject.next({ kind: JobOutboundMessageKind.ChannelError, description, name, error });
              // This can be reopened.
              channels.delete(name);
            },
            () => {
              subject.next({ kind: JobOutboundMessageKind.ChannelComplete, description, name });
              // This can be reopened.
              channels.delete(name);
            },
          );

          channels.set(name, channelSubject);

          return channelSubject;
        },
      };

      const result = fn(argument, newContext);
      // If the result is a promise, simply wait for it to complete before reporting the result.
      if (isPromise(result)) {
        result.then(result => {
          subject.next({ kind: JobOutboundMessageKind.Output, description, value: result });
          subject.next({ kind: JobOutboundMessageKind.End, description });
          subject.complete();
        }, err => subject.error(err));
      } else if (isObservable(result)) {
        subscription = (result as Observable<O>).subscribe(
          (value: O) => subject.next({ kind: JobOutboundMessageKind.Output, description, value }),
          error => subject.error(error),
          () => {
            subject.next({ kind: JobOutboundMessageKind.End, description });
            subject.complete();
          },
        );

        return subscription;
      } else {
        // If it's a scalar value, report it synchronously.
        subject.next({ kind: JobOutboundMessageKind.Output, description, value: result as O });
        subject.next({ kind: JobOutboundMessageKind.End, description });
        subject.complete();
      }
    });

    return new Observable<JobOutboundMessage<O>>(subject => {
      function complete() {
        if (subscription) {
          subscription.unsubscribe();
        }
        subject.next({ kind: JobOutboundMessageKind.End, description });
        subject.complete();
        inputChannel.complete();
      }

      // Handle input.
      const inboundSub = inboundBus.subscribe(message => {
        switch (message.kind) {
          case JobInboundMessageKind.Ping:
            subject.next({ kind: JobOutboundMessageKind.Pong, description, id: message.id });
            break;

          case JobInboundMessageKind.Stop:
            // There's no way to cancel a promise or a synchronous function, but we do cancel
            // observables where possible.
            complete();
            break;

          case JobInboundMessageKind.Input:
            inputChannel.next(message.value);
            break;
        }
      });

      // Execute the function with the additional context.
      const channels = new Map<string, Subject<JsonValue>>();

      const newContext: SimpleJobHandlerContext<A, I, O> = {
        ...context,
        input: inputChannel.asObservable(),
        createChannel(name: string) {
          if (channels.has(name)) {
            throw new ChannelAlreadyExistException(name);
          }
          const channelSubject = new Subject<JsonValue>();
          const channelSub = channelSubject.subscribe(
            message => {
              subject.next({
                kind: JobOutboundMessageKind.ChannelMessage, description, name, message,
              });
            },
            error => {
              subject.next({ kind: JobOutboundMessageKind.ChannelError, description, name, error });
              // This can be reopened.
              channels.delete(name);
            },
            () => {
              subject.next({ kind: JobOutboundMessageKind.ChannelComplete, description, name });
              // This can be reopened.
              channels.delete(name);
            },
          );

          channels.set(name, channelSubject);
          if (subscription) {
            subscription.add(channelSub);
          }

          return channelSubject;
        },
      };

      subject.next({ kind: JobOutboundMessageKind.Start, description });
      let result = fn(argument, newContext);
      // If the result is a promise, simply wait for it to complete before reporting the result.
      if (isPromise(result)) {
        result = from(result);
      } else if (!isObservable(result)) {
        result = of(result as O);
      }

      subscription = (result as Observable<O>).subscribe(
        (value: O) => subject.next({ kind: JobOutboundMessageKind.Output, description, value }),
        error => subject.error(error),
        () => complete(),
      );
      subscription.add(inboundSub);

      return subscription;
    });

//.........这里部分代码省略.........
            });

            // We don't want to subscribe errors and complete.
            subscriptions.push(run.progress.subscribe(event => progressChannel.next(event)));

            return run;
          },
          async scheduleBuilder(
            builderName: string,
            options: json.JsonObject = {},
            scheduleOptions: ScheduleOptions = {},
          ) {
            const run = await scheduleByName(builderName, options, {
              scheduler,
              logger: scheduleOptions.logger || logger.createChild(''),
              workspaceRoot: i.workspaceRoot,
              currentDirectory: i.currentDirectory,
            });

            // We don't want to subscribe errors and complete.
            subscriptions.push(run.progress.subscribe(event => progressChannel.next(event)));

            return run;
          },
          async getTargetOptions(target: Target) {
            return scheduler.schedule<Target, json.JsonValue, json.JsonObject>(
                    '..getTargetOptions', target).output.toPromise();
          },
          async getBuilderNameForTarget(target: Target) {
            return scheduler.schedule<Target, json.JsonValue, string>(
              '..getBuilderNameForTarget',
              target,
            ).output.toPromise();
          },
          async validateOptions<T extends json.JsonObject = json.JsonObject>(
            options: json.JsonObject,
            builderName: string,
          ) {
            return scheduler.schedule<[string, json.JsonObject], json.JsonValue, T>(
              '..validateOptions',
              [builderName, options],
            ).output.toPromise();
          },
          reportRunning() {
            switch (currentState) {
              case BuilderProgressState.Waiting:
              case BuilderProgressState.Stopped:
                progress({ state: BuilderProgressState.Running, current: 0, total  }, context);
                break;
            }
          },
          reportStatus(status: string) {
            switch (currentState) {
              case BuilderProgressState.Running:
                progress({ state: currentState, status, current, total  }, context);
                break;
              case BuilderProgressState.Waiting:
                progress({ state: currentState, status }, context);
                break;
            }
          },
          reportProgress(current: number, total?: number, status?: string) {
            switch (currentState) {
              case BuilderProgressState.Running:
                progress({ state: currentState, current, total, status }, context);
            }
          },
          analytics: new analytics.ForwardingAnalytics(report => analyticsChannel.next(report)),
          addTeardown(teardown: () => (Promise<void> | void)): void {
            teardownLogics.push(teardown);
          },
        };

        context.reportRunning();
        let result: BuilderOutputLike;
        try {
          result = fn(i.options as OptT, context);
        } catch (e) {
          result = throwError(e);
        }

        if (isPromise(result)) {
          result = from(result);
        } else if (!isObservable(result)) {
          result = of(result);
        }

        // Manage some state automatically.
        progress({ state: BuilderProgressState.Running, current: 0, total: 1 }, context);
        subscriptions.push(result.pipe(
          tap(() => {
            progress({ state: BuilderProgressState.Running, current: total }, context);
            progress({ state: BuilderProgressState.Stopped }, context);
          }),
        ).subscribe(
          message => observer.next(message as OutT),
          error => observer.error(error),
          () => observer.complete(),
        ));
      }