Method stream

packages/core/src/EntityManager.ts:334–385

* Finds all entities and returns an async iterable (async generator) that yields results one by one. * The results are merged and mapped to entity instances, without adding them to the identity map. * You can disable merging by passing the options `{ mergeResults: false }`. * With `mergeRes

(
    entityName: EntityName<Entity>,
    options: WithUsingOptions<
      StreamOptions<NoInfer<Entity>, Hint, Fields, Excludes>,
      NoInfer<Entity>,
      Using
    > = {} as any,
  )

   * ```
   */
  async *stream<
    Entity extends object,
    Hint extends string = never,
    Fields extends string = never,
    Excludes extends string = never,
    Using extends string = never,
  >(
    entityName: EntityName<Entity>,
    options: WithUsingOptions<
      StreamOptions<NoInfer<Entity>, Hint, Fields, Excludes>,
      NoInfer<Entity>,
      Using
    > = {} as any,
  ): AsyncIterableIterator<Loaded<Entity, Hint, Fields, Excludes>> {
    const em = this.getContext();
    em.prepareOptions(options);
    (options as Dictionary).strategy = 'joined';
    await em.tryFlush(entityName, options);
    const where = (await em.processWhere(entityName, options.where ?? {}, options, 'read')) as FilterQuery<Entity>;
    validateParams(where);
    options.orderBy = options.orderBy || {};
    options.populate = (await em.preparePopulate(entityName, options)) as any;
    const meta = this.metadata.get<Entity>(entityName);
    em.validateIndexUsage(meta, options.where ?? {}, options);
    options = { ...options };
    // save the original hint value so we know it was infer/all
    (options as Dictionary)._populateWhere = options.populateWhere ?? this.config.get('populateWhere');
    options.populateWhere = this.createPopulateWhere({ ...where } as ObjectQuery<Entity>, options);
    options.populateFilter = await this.getJoinedFilters(meta, options);
    const stream = em.driver.stream(entityName, where, {
      ctx: em.#transactionContext,
      mapResults: false,
      ...options,
    } as StreamOptions<Entity>);

    for await (const data of stream) {
      const fork = em.fork();
      const entity = fork.#entityFactory.create(entityName, data as EntityData<Entity>, {
        refresh: options.refresh,
        schema: options.schema,
        convertCustomTypes: true,
      }) as Loaded<Entity, Hint, Fields, Excludes>;
      helper(entity).setSerializationContext({
        populate: options.populate,
        fields: options.fields,
        exclude: options.exclude,
      } as any);
      await fork.#unitOfWork.dispatchOnLoadEvent();
      fork.clear();
      yield entity;
    }
  }

  /**
   * Finds all entities of given type, optionally matching the `where` condition provided in the `options` parameter.