fix(throttle): no longer emits more than necessary in sync/sync trailing case

spec/operators/throttle-spec.ts

@@ -4,7 +4,7 @@ import { throttle, mergeMap, mapTo, take } from 'rxjs/operators';
 import { of, concat, timer, Observable } from 'rxjs';
 
 /** @test {throttle} */
-describe('throttle operator', () =>  {
+describe('throttle', () =>  {
   it('should immediately emit the first value in each time window', () =>  {
     const e1 =   hot('-a-xy-----b--x--cxxx-|');
     const e1subs =   '^                    !';
@@ -21,6 +21,20 @@ describe('throttle operator', () =>  {
     expectSubscriptions(e2.subscriptions).toBe(e2subs);
   });
 
+  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'])
+  });
+
   it('should simply mirror the source if values are not emitted often enough', () =>  {
     const e1 =   hot('-a--------b-----c----|');
     const e1subs =   '^                    !';

src/internal/operators/throttle.ts

@@ -90,11 +90,16 @@ export function throttle<T>(
 
     const send = () => {
       if (hasValue) {
-        subscriber.next(sendValue!);
-        !isComplete && startThrottle(sendValue!);
+        // Ensure we clear out our value and hasValue flag
+        // before we emit, otherwise reentrant code can cause
+        // issues here.
+        hasValue = false;
+        const value = sendValue!;
+        sendValue = null;
+        // Emit the value.
+        subscriber.next(value);
+        !isComplete && startThrottle(value);
       }
-      hasValue = false;
-      sendValue = null;
     };
 
     source.subscribe(