Function handler

packages/angular_devkit/architect/src/architect.ts:66–133

(argument: json.JsonObject, context: JobHandlerContext)

  };

  function handler(argument: json.JsonObject, context: JobHandlerContext) {
    // Add input validation to the inbound bus.
    const inboundBusWithInputValidation = context.inboundBus.pipe(
      concatMap(async (message) => {
        if (message.kind === JobInboundMessageKind.Input) {
          const v = message.value as BuilderInput;
          const options = mergeOptions(baseOptions, v.options);

          // Validate v against the options schema.
          const validation = await registry.compile(info.optionSchema);
          const validationResult = await validation(options);
          const { data, success, errors } = validationResult;

          if (!success) {
            throw new json.schema.SchemaValidationException(errors);
          }

          return { ...message, value: { ...v, options: data } } as JobInboundMessage<BuilderInput>;
        } else {
          return message as JobInboundMessage<BuilderInput>;
        }
      }),
      // Using a share replay because the job might be synchronously sending input, but
      // asynchronously listening to it.
      shareReplay(1),
    );

    // Make an inboundBus that completes instead of erroring out.
    // We'll merge the errors into the output instead.
    const inboundBus = onErrorResumeNext(inboundBusWithInputValidation);

    const output = from(host.loadBuilder(info)).pipe(
      concatMap((builder) => {
        if (builder === null) {
          throw new Error(`Cannot load builder for builderInfo ${JSON.stringify(info, null, 2)}`);
        }

        return builder.handler(argument, { ...context, inboundBus }).pipe(
          map((output) => {
            if (output.kind === JobOutboundMessageKind.Output) {
              // Add target to it.
              return {
                ...output,
                value: {
                  ...output.value,
                  ...(target ? { target } : 0),
                } as unknown as json.JsonObject,
              };
            } else {
              return output;
            }
          }),
        );
      }),
      // Share subscriptions to the output, otherwise the handler will be re-run.
      shareReplay(),
    );