return subscription stubs if props not defined

Author: tusharmath

src/Observable.ts

@@ -29,7 +29,7 @@ export class Observable<T> implements IObservable<T> {
 
   safelyExecuteFunc (observer: IObserver<T>, cSub: CompositeSubscription) {
     const r = SafeExecutor(() => {
-      cSub.add(Subscription.from(this.func(SubscriptionObserver.of(observer))))
+      cSub.add(Subscription.from(this.func(SubscriptionObserver.from(observer))))
     })
     if (r.type === Safety.error && observer.error) {
       observer.error(r.value as Error)

src/SubscriptionObserver.ts

@@ -5,11 +5,24 @@
 import {ISubscriptionObserver} from './types/core/ISubscriptionObserver';
 import {IObserver} from './types/core/IObserver';
 
+export class SubscriptionObserverStub <T> implements ISubscriptionObserver<T> {
+  closed: boolean;
+
+  next (val: T): void {
+  }
+
+  error (err: Error): void {
+  }
+
+  complete (): void {
+  }
+}
+
 export class SubscriptionObserver<T> implements ISubscriptionObserver<T> {
   closed: boolean;
 
   constructor (private sink: IObserver<T>) {
-    this.closed = !this.sink.next
+    this.closed = false
   }
 
   next (val: T): void {
@@ -18,15 +31,16 @@ export class SubscriptionObserver<T> implements ISubscriptionObserver<T> {
 
   error (err: Error): void {
     this.sink.error(err)
-
   }
 
   complete (): void {
-    if (this.sink.complete) this.sink.complete()
+    this.sink.complete()
     this.closed = true
   }
 
-  static of<T> (observer: IObserver<T>) {
+  static from<T> (observer: IObserver<T>) {
+    if (!observer.next || !observer.complete || !observer.error)
+      return new SubscriptionObserverStub()
     return new SubscriptionObserver(observer)
   }
 }