run method

Implementation

Future<RunResult<Output>> run(
  Input? input, {
  StreamingCallback<Chunk>? onChunk,
  Map<String, dynamic>? context,
  Stream<Input>? inputStream,
  Init? init,
  TraceStartCallback? onTraceStart,
  CancellationToken? cancel,
}) async {
  // Bail before doing any work if the caller's token is already cancelled.
  cancel?.throwIfCancelled();

  if (inputStream == null) {
    final internalInputController = StreamController<Input>();
    inputStream = internalInputController.stream;
    if (input != null) {
      internalInputController.add(input);
    }
    internalInputController.close();
  }

  final executionContext = context ?? Zone.current[_genkitContextKey];
  Future<RunResult<Output>> runner() async {
    var traceId = '';
    var spanId = '';
    final result = await runInNewSpan(
      name,
      (telemetryContext) async {
        traceId = telemetryContext.traceId;
        spanId = telemetryContext.spanId;
        if (onTraceStart != null) {
          onTraceStart(traceId: traceId, spanId: spanId);
        }
        _recordContextMetadata(executionContext);
        return await fn(input, (
          streamingRequested: onChunk != null,
          sendChunk: onChunk ?? (chunk) {},
          context: executionContext,
          inputStream: inputStream,
          init: init,
          cancel: cancel,
        ));
      },
      actionType: actionType.value,
      input: input,
    );
    return RunResult<Output>(
      result: result,
      traceId: traceId,
      spanId: spanId,
    );
  }

  if (context != null) {
    return runZoned(runner, zoneValues: {_genkitContextKey: context});
  } else {
    return runner();
  }
}