fix(throttleTime): emit throttled values when complete if trailing=tr…

…ue (#3559)

Treat complete as a trailing edge and emit the last throttled value when complete before the throttle time is up.

closes #3351

spec/operators/throttleTime-spec.ts

@@ -164,5 +164,17 @@ describe('Observable.prototype.throttleTime', () => {
       expectObservable(result).toBe(expected);
       expectSubscriptions(e1.subscriptions).toBe(e1subs);
     });
+
+    asDiagram('throttleTime(fn, { leading: false, trailing: true })')('should emit the last throttled value when complete', () => {
+      const e1 =   hot('-a-xy-----b--x--cxx|');
+      const e1subs =   '^                  !';
+      const t =   time('----|              ');
+      const expected = '-----y--------x----(x|)';
+
+      const result = e1.throttleTime(t, rxTestScheduler, { leading: false, trailing: true });
+
+      expectObservable(result).toBe(expected);
+      expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    });
   });
 });

src/internal/operators/throttleTime.ts

@@ -97,6 +97,15 @@ class ThrottleTimeSubscriber<T> extends Subscriber<T> {
     }
   }
 
+  protected _complete() {
+    if (this._hasTrailingValue) {
+      this.destination.next(this._trailingValue);
+      this.destination.complete();
+    } else {
+      this.destination.complete();
+    }
+  }
+
   clearThrottle() {
     const throttled = this.throttled;
     if (throttled) {