Feature Request: resetOnNext and resetOn support for share

[pcbowers]

Motivation

Having the ability to reset a share operator arbitrarily based on emitted values or external triggers would be incredibly useful, especially in scenarios like:

• Cache invalidation
• Trigger-based refreshes
• Handling sentinel values (e.g. { type: 'reset' })
• Time or user driven resets, regardless of emissions

The existing reset options (resetOnError, resetOnComplete, and resetOnRefCountZero) are tied to lifecycle events, but there's currently no mechanism to reset based on next emissions or external signals.

Proposed API

Extend ShareConfig<T> with:

resetOnNext?: false | ((value: T) => ObservableInput<any>);
resetOn?: false | ObservableInput<any>;

Notes

• Allowing false for both resetOnNext and resetOn gives us the ability to have the defaults for both be false instead of undefined
• resetOnNext intentionally does not accept true, because unconditional reset on every emission defeats the purpose of sharing.
• Same can be said for resetOn since a truthy value doesn't make sense: we need some value to trigger on
• The feature aligns well with existing handleReset() infrastructure and offers significant expressive power with minimal added complexity.
• These are non-breaking additions that expand the expressive power of share without affecting current usage.

Examples

Reset on specific value:

source$.pipe(
  share({
    resetOnNext: value => (value === 'invalidate' ? of(null) : NEVER)
  })
);

Reset with delay after special value:

source$.pipe(
  share({
    resetOnNext: value => (value?.type === 'refresh' ? timer(5000) : NEVER)
  })
);

Reset on internally handled errors

source$.pipe(
  switchMap(query => fetch$(query).pipe(catchError(error => of({ type: 'error', error })))),
  share({
    resetOnNext: value => (value.type === 'error' ? of(null) : NEVER)
  })
);

Time-based resets

source$.pipe(
  share({
    resetOn: interval(30_000)
  })
);

Manual resets via Subject

const reset$ = new Subject<void>();

source$.pipe(
  share({
    resetOn: reset$
  })
);

// Somewhere else:
reset$.next(); // triggers reset

Implementation Sketch

// packages/rxjs/src/internal/operators/share.ts

export interface ShareConfig<T> {
  ...
  /**
   * A factory function that receives each value emitted by the source observable and returns an `ObservableInput` that
   * controls when the internal state should be reset. This allows the resulting observable to be reset based on arbitrary
   * emissions — such as specific flags, conditions, or signals carried within the emitted values.
   */
  resetOnNext?: (value: T) => ObservableInput<any>;
  /**
   * An external ObservableInput that, when it emits, will trigger a reset
   * of the internal state regardless of the source's emissions or lifecycle.
   */
  resetOn?: ObservableInput<any>;
}

if (resetOn) {
  subscriber.add(from(resetOn).subscribe(reset));
}

connection = new SafeSubscriber({
  next: (value: T) => {
    dest.next(value);
    cancelReset();
    resetConnection = handleReset(reset, resetOnNext, value);
  },
  ...
});

This follows the same pattern already used by resetOnError and resetOnComplete.