RxJS wrapper for DataScript. DataScript connection represented as reports stream and transactions stream. Reports stream is formed from transactions stream with using scan function.
import {datascript, mori, helpers} from 'datascript-mori'
import {connect, nextTx, q$, entity$} from '../src/index'
import 'rxjs/add/operator/skipWhile'
import 'rxjs/add/operator/filter'
const {DB_ID, DB_ADD, TX_DATA, TX_META, DB_AFTER, DB_BEFORE, DB_UNIQUE, DB_UNIQUE_IDENTITY} = helpers
const {vector, parse, get, hashMap, map, nth, reduce} = mori
const {js: djs} = datascript
const db = djs.empty_db({name: {[DB_UNIQUE]: DB_UNIQUE_IDENTITY}})
const {report$, tx$} = connect(db) // connect is a stream of transactions and stream of reports
const ivanAdultEntity$ = report$
::entity$(vector(`name`, `Ivan`)) // make entity stream
.skipWhile(
Ivan => get(Ivan, `age`) < 18
) // skip all entity with age < 18
const names$ = report$
.filter(
report => find(
map(report, tx => nth(tx, 2)),
`name`
)
) // filter all tx which dont affect names of entities
::q$(parse(`[:find [?n ...] :where [?e "name" ?n]]`)) // make results of the query stream
// subscribes
names$.subscribe(
names => console.log(
`Names of users: ${reduce(names, (acc, name) => acc + ', ' + name)}`
)
)
ivanAdultEntity$.subscribe(Ivan => console.log(`Ivan age ${get(Ivan, 'age')} years`))
// Add some tx
nextTx(tx$, vector(
vector(DB_ADD, 1, `name`, `Ivan`),
vector(DB_ADD, 1, `age`, 17)
))
nextTx(tx$, vector(
vector(DB_ADD, 1, `age`, 18)
))
nextTx(tx$, vector(
vector(DB_ADD, 1, `age`, 19)
))
nextTx(tx$, vector(
hashMap(
DB_ID, 2,
"name", "Igor",
"age", 35
)
));
/* Output entity subscriber
Ivan age 18 years
Ivan age 19 years
*/
/* Output names subscriber
Names of users: Ivan
Names of users: Ivan, Igor
*/createAnyQueryStream(queryFunc: Function, distinctUntilChangedFunc: Function): Function: Basic high order function for wrapping any query function in stream. Second optional argument is function that compare results of query.connect(db: DataScript DB): {report$: Observable<Report>, tx$: Observable<tx>}: Function for creating observable of report and tx from DataScript DB.nextTx(tx$: Observable<tx>, ...tx: Array<tx>):Void: Function for applying array of transactions to stream of tx.
q$entity$filter$pull$pullMany$datoms$seekDatoms$indexRange$
All functions takes the first argument observable of reports. Also you can pass observable of reports in this with using bind operator - report$::q$(...). Rest arguments is equals list of arguments in original function(see DataScript Docs)
DataScript transaction API based on callbacks. Callbacks is a bad, because using callbacks, you can not work with events as first-class citizen value. Reactive streams allow the manipulate events as first-class citizen values and provide many operators for proccessing event streams.
- Interoperability with other streaming libs(Most, Kefir, etc.)