rethinkdb-observable

Convert a rethinkdb cursor into an observable

Installation

npm i --save rethinkdb-observable
npm i --save rxjs # peer dependency

Usage

Example: observable w/ single subscribe/unsubscribe

var createObservable = require('rethinkdb-observable')
var r = require('rethinkdb')
 
rethinkdb.table('test').run(conn).then(function (cursor) {
  // Note: this is a basic observable and only allows ONE subscription. for multiple, see example below.
  var observable = createObservable(cursor)
  // subscribe usage
  var subscription = observable.subscribe(
    function onNext (next) {
      // onNext will be passed each item as they are recieved from the cursor
    },
    function onError (err) {
      // onError will trigger for any cursor errors
    },
    function onCompleted () {
      // on complete will trigger after last "next" has been pushed
      // and cursor has closed successfully
    }
  )
  // unsubscribe usage
  subscription.unsubscribe()
  // unsubscribe will detach the subscription callbacks and close the cursor
})

Example: observable w/ multiple subscriptions

Uses rxjs ConnectableObservable by using publish. To learn more about ReactiveX observables checkout: reactivex.io or intro to rx

var createObservable = require('rethinkdb-observable')
var r = require('rethinkdb')
// required to use publish
require('rxjs/add/operator/publish')
 
rethinkdb.table('test').run(conn).then(function (cursor) {
  // Note: this is a basic observable and only allows ONE subscription. for multiple, see example below.
  var observable = rethinkdbObservable(cursor).publish().refCount()
  // subscribe usage
  var subscription = observable.subscribe(
    function onNext (next) {
      // onNext will be passed each item as they are recieved from the cursor
    },
    function onError (err) {
      // onError will trigger for any cursor errors
    },
    function onCompleted () {
      // on complete will trigger after last "next" has been pushed
      // and cursor has closed successfully
    }
  )
  // unsubscribe usage
  subscription.unsubscribe()
  // unsubscribe will detach the subscription callbacks and close the cursor
})

License

MIT