hunk.code_window:     expectObservable(result).toBe(expected);
       expectSubscriptions(e1.subscriptions).toBe(e1subs);
       expectSubscriptions(e2.subscriptions).toBe(e2subs);
     });
     it('should simply mirror the source if values are not emitted often enough', () =>  {
       const e1 =   hot('-a--------b-----c----|');
       const e1subs =   '^                    !';
       const e2 =  cold( '----x                ');
  
hunk.labels: keep keep keep keep add keep keep keep keep
hunk.after_edit:   it('should handle sync source with sync notifier and trailing appropriately', () => {
       let results: any[] = [];
       const source = of(1).pipe(
         throttle(() => of(1), { leading: false, trailing: true })
       );
       source.subscribe({
         next: value => results.push(value),
         complete: () => results.push('done')
       });
       expect(results).to.deep.equal([1, 'done'])
     });
  
hunk.file_path: spec/operators/throttle-spec.ts
hunk.type: add
file: import { Subscription } from '../Subscription';
  
  import { MonoTypeOperatorFunction, ObservableInput } from '../types';
  import { operate } from '../util/lift';
  import { OperatorSubscriber } from './OperatorSubscriber';
  import { innerFrom } from '../observable/from';
  
  export interface ThrottleConfig {
    leading?: boolean;
    trailing?: boolean;
  }
  
  export const defaultThrottleConfig: ThrottleConfig = {
    leading: true,
    trailing: false,
  };
  
  /**
   * Emits a value from the source Observable, then ignores subsequent source
   * values for a duration determined by another Observable, then repeats this
   * process.
   *
   * <span class="informal">It's like {@link throttleTime}, but the silencing
   * duration is determined by a second Observable.</span>
   *
   * ![](throttle.png)
   *
   * `throttle` emits the source Observable values on the output Observable
   * when its internal timer is disabled, and ignores source values when the timer
   * is enabled. Initially, the timer is disabled. As soon as the first source
   * value arrives, it is forwarded to the output Observable, and then the timer
   * is enabled by calling the `durationSelector` function with the source value,
   * which returns the "duration" Observable. When the duration Observable emits a
   * value, the timer is disabled, and this process repeats for the
   * next source value.
   *
   * ## Example
   * Emit clicks at a rate of at most one click per second
   * ```ts
   * import { fromEvent, interval } from 'rxjs';
   * import { throttle } from 'rxjs/operators';
   *
   * const clicks = fromEvent(document, 'click');
   * const result = clicks.pipe(throttle(ev => interval(1000)));
   * result.subscribe(x => console.log(x));
   * ```
   *
   * @see {@link audit}
   * @see {@link debounce}
   * @see {@link delayWhen}
   * @see {@link sample}
   * @see {@link throttleTime}
   *
   * @param durationSelector A function
   * that receives a value from the source Observable, for computing the silencing
   * duration for each source value, returned as an Observable or a Promise.
   * @param config a configuration object to define `leading` and `trailing` behavior. Defaults
   * to `{ leading: true, trailing: false }`.
   * @return An Observable that performs the throttle operation to
   * limit the rate of emissions from the source.
   */
  export function throttle<T>(
    durationSelector: (value: T) => ObservableInput<any>,
    { leading, trailing }: ThrottleConfig = defaultThrottleConfig
  ): MonoTypeOperatorFunction<T> {
    return operate((source, subscriber) => {
      let hasValue = false;
      let sendValue: T | null = null;
      let throttled: Subscription | null = null;
      let isComplete = false;
  
      const endThrottling = () => {
        throttled?.unsubscribe();
        throttled = null;
        if (trailing) {
          send();
          isComplete && subscriber.complete();
        }
      };
  
      const cleanupThrottling = () => {
        throttled = null;
        isComplete && subscriber.complete();
      };
  
      const startThrottle = (value: T) =>
        (throttled = innerFrom(durationSelector(value)).subscribe(
          new OperatorSubscriber(subscriber, endThrottling, undefined, cleanupThrottling)
        ));
  
      const send = () => {
        if (hasValue) {
          subscriber.next(sendValue!);
          !isComplete && startThrottle(sendValue!);
        }
        hasValue = false;
        sendValue = null;
      };
  
      source.subscribe(
        new OperatorSubscriber(
          subscriber,
          // Regarding the presence of throttled.closed in the following
          // conditions, if a synchronous duration selector is specified - weird,
          // but legal - an already-closed subscription will be assigned to
          // throttled, so the subscription's closed property needs to be checked,
          // too.
          (value) => {
            hasValue = true;
            sendValue = value;
            !(throttled && !throttled.closed) && (leading ? send() : startThrottle(value));
          },
          undefined,
          () => {
            isComplete = true;
            !(trailing && hasValue && throttled && !throttled.closed) && subscriber.complete();
          }
        )
      );
    });
  }
  