feat(main): control fetch() based on subscription count

instead of using @asStream() to determine  the state of the component and thus figure out if the
HTTP request needs to be made or not, we decide it based on if there is a subscription or not

BREAKING CHANGES: @asStream() is not needed. removed reload(). removed isLoading().

index.js

@@ -5,7 +5,9 @@
 
 const create = require('./src/main').create
 
-exports.create = (request, fetch) => create(fetch || window.fetch, request)
+exports.create = (request, fetch) => create(
+  fetch || window.fetch.bind(window),
+  request
+)
+
 exports.toJSON = require('./src/toJSON')
-exports.reload = require('./src/reload')
-exports.isLoading = require('./src/isLoading')

package.json

@@ -9,7 +9,6 @@
   "author": "tusharmath@gmail.com",
   "license": "ISC",
   "dependencies": {
-    "argtoob": "^1.0.2",
     "funjector": "^3.1.0"
   },
   "peerDependencies": {
@@ -21,7 +20,8 @@
     "cz-conventional-changelog": "^1.1.5",
     "nyc": "^6.0.0",
     "rx": "^4.1.0",
-    "semantic-release": "^4.3.5"
+    "semantic-release": "^4.3.5",
+    "sinon": "^1.17.3"
   },
   "repository": {
     "type": "git",

src/main.js

@@ -4,49 +4,20 @@
 'use strict'
 
 const Rx = require('rx')
-const targs = require('argtoob')
 const _ = require('funjector')
 const e = exports
-
-e.create = _.partial((xhr, fetch, request) => {
-  const observer = new Rx.Subject()
-  xhr(fetch, request, observer).subscribe(observer)
-  return observer
-}, e.xhr)
-
-e.xhr = (fetch, request, observer) => Rx.Observable.merge(
-  e.fetchBegin(request, observer),
-  e.fetchEnd(fetch, observer)
-)
-
-e.isHydrated = events => {
-  const mount = events.filter(x => x.event === 'WILL_MOUNT').map(1)
-  const unmount = events.filter(x => x.event === 'WILL_UNMOUNT').map(-1)
-  return Rx.Observable
-    .merge(mount, unmount).startWith(0)
-    .scan((m, x) => x + m, 0)
-    .map(x => x > 0)
-    .distinctUntilChanged()
+const slice = (x, i) => Array.prototype.slice.call(x, i)
+
+e.create = _.partial((toObservable, fetch, request) => {
+  const fetchO = toObservable(fetch)
+  const out = request
+    .flatMap(x => fetchO.apply(null, x))
+    .replay(1)
+  out.connect()
+  return out
+}, e.toObservable)
+
+e.toObservable = fetch => function () {
+  const args = slice(arguments)
+  return Rx.Observable.fromPromise(fetch.apply(null, args))
 }
-
-e.getHydratedRequests = _.partial((isHydrated, request, com) => request
-  .combineLatest(isHydrated(com), targs('request', 'isHydrated'))
-  .filter(x => x.isHydrated)
-  .pluck('request'),
-  e.isHydrated
-)
-
-e.fetchBegin = _.partial((getHydratedRequests, req, com) => {
-  const reload = com
-    .filter(x => x.event === 'RELOAD')
-    .startWith(null)
-
-  return getHydratedRequests(req, com)
-    .map(x => ({event: 'REQUEST', args: x}))
-    .combineLatest(reload, a => a)
-}, e.getHydratedRequests)
-
-e.fetchEnd = (fetch, com) => com
-  .filter(x => x.event === 'REQUEST')
-  .flatMap(x => fetch.apply(null, x.args))
-  .map(x => ({event: 'RESPONSE', args: [x]}))

src/toJSON.js

@@ -1,18 +1,5 @@
-const Rx = require('rx')
-const targs = require('argtoob')
-module.exports = store => {
-  const request = store
-    .filter(x => x.event === 'REQUEST').pluck('args')
+'use strict'
 
-  const response = store
-    .filter(x => x.event === 'RESPONSE').map(x => x.args[0])
-
-  const json = response.flatMap(x => x.json())
-
-  return Rx.Observable.zip(
-    request,
-    response,
-    json,
-    targs('request', 'response', 'json')
-  ).share()
-}
+module.exports = store => store.flatmap(
+  response => response.json().then(json => ({json, response}))
+)

test/test.main.js

@@ -3,82 +3,53 @@
  */
 
 'use strict'
-import Rx, {TestScheduler, ReactiveTest} from 'rx'
+import {spy} from 'sinon'
+import {TestScheduler, ReactiveTest} from 'rx'
 import test from 'ava'
-import _ from 'funjector'
-import {create, xhr} from '../src/main'
-import reload from '../src/reload'
+import {create} from '../src/main'
+import {call} from 'funjector'
 const {onNext, onCompleted} = ReactiveTest
+const identity = x => x
 
-test('returns subject', t => {
+test('calls fetch', t => {
+  const fetch = spy(identity)
+  const toObservable = () => fetch
   const sh = new TestScheduler()
-  const fetch = () => sh.createHotObservable()
-  t.true(_.call(create, fetch) instanceof Rx.Subject)
+  const request = sh.createHotObservable(
+    onNext(210, ['a', 1]),
+    onNext(220, ['b', 2])
+  )
+  call(create, toObservable, null, request)
+    .subscribe(identity)
+  sh.start()
+  t.true(fetch.getCall(0).calledWith('a', 1))
+  t.true(fetch.getCall(1).calledWith('b', 2))
 })
 
-test(t => {
-  const out = []
+test('multiple subscriptions', t => {
+  const fetch = spy(identity)
+  const toObservable = () => fetch
   const sh = new TestScheduler()
-  const fetch = (url, options) => sh.createHotObservable(
-    onNext(230, url + options.a)
-  )
   const request = sh.createHotObservable(
-    onNext(210, ['/a', {a: 0}])
+    onNext(210, ['a', 1]),
+    onNext(220, ['b', 2])
   )
-  const subject = _.call(create, xhr, fetch, request)
-
-  subject.subscribe(x => out.push(x))
-  subject.onNext({event: 'WILL_MOUNT'})
-
+  const store = call(create, toObservable, null, request)
+  store.subscribe(identity)
+  store.subscribe(identity)
   sh.start()
-  subject.onNext({event: 'WILL_MOUNT'})
-  t.same(out, [
-    {event: 'WILL_MOUNT'},
-    {event: 'REQUEST', args: ['/a', {a: 0}]},
-    {event: 'RESPONSE', args: ['/a0']},
-    {event: 'WILL_MOUNT'}
-  ])
+  t.is(fetch.callCount, 2)
 })
 
-test('reload', t => {
-  const out = []
+test('must get last value', t => {
+  const fetch = spy(x => x)
+  const toObservable = () => fetch
   const sh = new TestScheduler()
-  const fetches = [
-    sh.createHotObservable(
-      onNext(230, 'resp-230'),
-      onCompleted(231)
-    ),
-    sh.createHotObservable(
-      onNext(240, 'resp-240'),
-      onCompleted(241)
-    ),
-    sh.createHotObservable(
-      onNext(250, 'resp-250'),
-      onCompleted(251)
-    )
-  ]
-  const fetch = (i) => fetches.shift()
   const request = sh.createHotObservable(
-    onNext(210, ['/x'])
+    onNext(100, ['a', 1]),
+    onNext(110, ['b', 2])
   )
-  const subject = _.call(create, xhr, fetch, request)
-
-  subject.subscribe(x => out.push(x))
-  subject.onNext({event: 'WILL_MOUNT'})
-  sh.advanceTo(230)
-  reload(subject)
-  sh.advanceTo(240)
-  reload(subject)
-  sh.advanceTo(250)
-  t.same(out, [
-    {event: 'WILL_MOUNT'},
-    {event: 'REQUEST', args: ['/x']},
-    {event: 'RESPONSE', args: ['resp-230']},
-    {event: 'REQUEST', args: ['/x']},
-    {event: 'RELOAD', args: []},
-    {event: 'RESPONSE', args: ['resp-240']},
-    {event: 'REQUEST', args: ['/x']},
-    {event: 'RELOAD', args: []},
-    {event: 'RESPONSE', args: ['resp-250']}
-  ])
+  sh.startScheduler(() => call(create, toObservable, null, request))
+  t.is(fetch.callCount, 1)
+  t.true(fetch.calledWith('b', 2))
 })