<commit_before>import { Observable } from 'rxjs';
import { classToPlain, Expose, plainToClassFromExist, serialize } from 'class-transformer';
import { AttemptOptions, retry } from '@lifeomic/attempt';
import * as Chance from 'chance';
import { IgResponseError } from '../errors';
import { Repository } from './repository';
import { Enumerable } from '../decorators';

export abstract class Feed<Response = any, Item = any> extends Repository {
  public attemptOptions: Partial<AttemptOptions<any>> = {
    delay: 60000,
    factor: 1.5,
    maxAttempts: 10,
    minDelay: 60000,
    maxDelay: 300000,
    jitter: true,
  };
  public items$ = new Observable<Item[]>(observer => {
    let subscribed = true;
    process.nextTick(async () => {
      do {
        try {
          await retry(
            async () => {
              const items = await this.items();
              observer.next(items);
            },
            {
              handleError(error, context) {
                // If instagram just tells us to wait - we are waiting.
                if (
                  error instanceof IgResponseError &&
                  error.response.statusCode === 400 &&
                  error.response.body.status === 'fail'
                ) {
                  return;
                } else {
                  context.abort();
                }
              },
              ...this.attemptOptions,
            },
          );
        } catch (e) {
          observer.error(e);
        }
      } while (this.isMoreAvailable() && subscribed);
      observer.complete();
    });
    return () => (subscribed = false);
  });
  @Expose()
  protected moreAvailable: boolean;
  @Enumerable(false)
  protected chance = new Chance();<commit_msg>(feed) Implement configurable .observable() with semaphore<commit_after>import { Observable } from 'rxjs';
import { classToPlain, Expose, plainToClassFromExist, serialize } from 'class-transformer';
import { AttemptOptions, retry } from '@lifeomic/attempt';
import * as Chance from 'chance';
import { IgResponseError } from '../errors';
import { Repository } from './repository';
import { Enumerable } from '../decorators';

export abstract class Feed<Response = any, Item = any> extends Repository {
  public attemptOptions: Partial<AttemptOptions<any>> = {
    delay: 60000,
    factor: 1.5,
    maxAttempts: 10,
    minDelay: 60000,
    maxDelay: 300000,
    jitter: true,
  };
  public get items$() {
    return this.observable();
  }
  public observable(semaphore?: () => Promise<any>, attemptOptions?: Partial<AttemptOptions<any>>) {
    return new Observable<Item[]>(observer => {
      let subscribed = true;
      process.nextTick(async () => {
        do {
          try {
            await retry(
              async () => {
                const items = await this.items();
                observer.next(items);
                if (typeof semaphore === 'function') {
                  await semaphore();
                }
              },
              {
                handleError(error, context) {
                  // If instagram just tells us to wait - we are waiting.
                  if (
                    error instanceof IgResponseError &&
                    [400, 429, 500, 502].includes(error.response.statusCode) &&
                    subscribed
                  ) {
                    return;
                  } else {
                    context.abort();
                  }
                },
                ...(attemptOptions || this.attemptOptions),
              },
            );
          } catch (e) {
            observer.error(e);
          }
        } while (this.isMoreAvailable() && subscribed);
        observer.complete();
      });
      return function unsubscribe() {
        subscribed = false;
      };
    });
  }
  @Expose()
  protected moreAvailable: boolean;
  @Enumerable(false)
  protected chance = new Chance();
{'fraction_non_alphanumeric': 0.12012241775057383, 'fraction_numerical': 0.014027033919918388, 'mean_word_length': 1.746498599439776, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 22, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2708591', 'n_tokens_mistral': 1094, 'n_tokens_neox': 998, 'n_words': 259}