Merge pull request #230 from tusharmath/support-subject

Support subject

Author: tusharmath

API.md

@@ -14,11 +14,12 @@
   - [frames](#frames)
   - [fromArray](#fromArray)
   - [fromDOM](#fromDOM)
+  - [fromNodeStream](#fromNodeStream)
   - [fromPromise](#fromPromise)
   - [interval](#interval)
   - [just](#just)
   - [never](#never)
-  - [fromNodeStream](#fromNodeStream)
+  - [subject](#subject)
 
 - [Operators](#operators)
   - [combine](#combine)
@@ -333,6 +334,23 @@ O.forEach(console.log, $) // logs `AIR`
 function never(): Observable
 ```
 
+## subject
+
+```ts
+function subject(): Observable & Observer
+```
+`Subject` is a special type that is both and `Observer` and also an `Observable`.
+
+**Example:**
+```ts
+const $ = O.subject()
+
+O.forEach(console.log, $) // logs A
+
+$.next('A')
+
+```
+
 ## fromNodeStream
 
 ```ts
@@ -694,6 +712,9 @@ interface TestScheduler extends Scheduler {
 
   // Creates a TestObserver. TestObserver keeps log of when and what type of an event was fired.
   Observer (): Observer
+
+  // Schedules multiple jobs using an array of tasks
+  timeline (tasks: [[Number, () => void]]): void
 }
 ```
 

src/main.ts

@@ -35,6 +35,7 @@ export {sample} from './operators/Sample'
 export {scan} from './operators/Scan'
 export {skipRepeats} from './operators/SkipRepeats'
 export {slice} from './operators/Slice'
+export {subject} from './sources/Subject'
 export {switchLatest} from './operators/Switch'
 export {switchMap} from './operators/Switch'
 export {tap} from './operators/Map'

src/sources/Subject.ts

@@ -0,0 +1,27 @@
+/**
+ * Created by tushar on 11/01/18.
+ */
+import {LinkedListNode} from '../lib/LinkedList'
+import {IObservable} from '../lib/Observable'
+import {CompositeObserver, IObserver} from '../lib/Observer'
+import {IScheduler} from '../lib/Scheduler'
+import {ISubscription} from '../lib/Subscription'
+
+class SubjectSubscription<T> implements ISubscription {
+  constructor(
+    private cObserver: CompositeObserver<T>,
+    private node: LinkedListNode<IObserver<T>>
+  ) {}
+  closed: boolean
+
+  unsubscribe(): void {
+    this.cObserver.remove(this.node)
+  }
+}
+
+export class Subject<T> extends CompositeObserver<T> implements IObservable<T> {
+  subscribe(observer: IObserver<T>, scheduler: IScheduler): ISubscription {
+    return new SubjectSubscription(this, this.add(observer))
+  }
+}
+export const subject = <T>() => new Subject<T>()

src/testing/TestScheduler.ts

@@ -164,6 +164,10 @@ export class TestScheduler implements IScheduler {
   Observer<T>() {
     return new TestObserver<T>(this)
   }
+
+  timeline(tasks: Array<[number, () => void]>) {
+    tasks.forEach(([time, task]) => this.delay(task, time))
+  }
 }
 
 export const createTestScheduler = (

test/test.Subject.ts

@@ -0,0 +1,95 @@
+/**
+ * Created by tushar on 11/01/18.
+ */
+import * as assert from 'assert'
+import {ISubscription} from '../src/lib/Subscription'
+import {subject} from '../src/sources/Subject'
+import {EVENT} from '../src/testing/Events'
+import {createTestScheduler} from '../src/testing/TestScheduler'
+
+describe('subject()', () => {
+  it('should be an observable+observer', () => {
+    const sh = createTestScheduler()
+    const ob = sh.Observer()
+    const $ = subject()
+    let sub: ISubscription
+
+    // Timeline
+    sh.timeline([
+      [210, () => $.next('A')],
+      [220, () => (sub = $.subscribe(ob, sh))],
+      [230, () => $.next('B')],
+      [240, () => $.next('C')],
+      [250, () => sub.unsubscribe()],
+      [260, () => $.next('D')],
+      [270, () => $.complete()]
+    ])
+
+    sh.advanceTo(211)
+    assert.deepEqual(ob.results, []) // nothing gets captured
+
+    sh.advanceTo(241)
+    assert.deepEqual(ob.results, [EVENT.next(230, 'B'), EVENT.next(240, 'C')])
+
+    sh.advanceTo(261)
+    assert.deepEqual(ob.results, [EVENT.next(230, 'B'), EVENT.next(240, 'C')])
+
+    sh.advanceTo(271)
+    assert.deepEqual(ob.results, [EVENT.next(230, 'B'), EVENT.next(240, 'C')])
+  })
+
+  it('should be an observable+observer', () => {
+    const sh = createTestScheduler()
+    const ob0 = sh.Observer()
+    const ob1 = sh.Observer()
+    const $ = subject()
+    let sb1: ISubscription
+    let sb0: ISubscription
+
+    // Timeline
+    sh.timeline([
+      [210, () => $.next('A')],
+      // -- 211
+      [220, () => (sb0 = $.subscribe(ob0, sh))],
+      [230, () => $.next('B')],
+      // -- 231
+      [240, () => (sb1 = $.subscribe(ob1, sh))],
+      [250, () => $.next('C')],
+      // -- 251
+      [260, () => sb1.unsubscribe()],
+      [270, () => $.next('D')],
+      // -- 271
+      [280, () => sb0.unsubscribe()],
+      [290, () => $.next('E')]
+      // -- 291
+    ])
+
+    sh.advanceTo(211)
+    assert.deepEqual(ob0.results, [])
+    assert.deepEqual(ob1.results, [])
+
+    sh.advanceTo(231)
+    assert.deepEqual(ob0.results, [EVENT.next(230, 'B')])
+    assert.deepEqual(ob1.results, [])
+
+    sh.advanceTo(251)
+    assert.deepEqual(ob0.results, [EVENT.next(230, 'B'), EVENT.next(250, 'C')])
+    assert.deepEqual(ob1.results, [EVENT.next(250, 'C')])
+
+    sh.advanceTo(271)
+    assert.deepEqual(ob0.results, [
+      EVENT.next(230, 'B'),
+      EVENT.next(250, 'C'),
+      EVENT.next(270, 'D')
+    ])
+    assert.deepEqual(ob1.results, [EVENT.next(250, 'C')])
+
+    sh.advanceTo(291)
+    assert.deepEqual(ob0.results, [
+      EVENT.next(230, 'B'),
+      EVENT.next(250, 'C'),
+      EVENT.next(270, 'D')
+    ])
+    assert.deepEqual(ob1.results, [EVENT.next(250, 'C')])
+  })
+})