medium-rxjs-subsequent-composed.ts

function subsequent<T>(
  count: number,
  operator: (source: Observable<T>) => Observable<T>
): (source: Observable<T>) => Observable<T> {
  return (source: Observable<T>) => new Observable<T>(observer => {
      const published = source.pipe(publish()) as ConnectableObservable<T>;
      const concatenated = concat(
        published.pipe(take(count)),
        published.pipe(operator)
      );
      const subscription = concatenated.subscribe(observer);
      subscription.add(published.connect());
      return subscription;
    }
  );
}