@x/observable

Core observable functionality for libraries on the @x platform.

@x/observable contains the core observable types and utility functions that are used throughout libraries on the @x platform, including @x/expressions, @x/socket and @x/unify.

@x observables differ from many existing observable implementations in a few key ways:

• observables are always a distinct object themselves - no automatic proxying is performed. This allows observables to consistently be passed as arguments to functions, even if they are primitive values (strings, numbers, etc.)
• observables always encapsulate a "current value" that can be revealed using the currentValue function, or executing the observable object itself as a function
• related to the previous point, observables do not explicitly have an "end" before the result can be interrogated; input event streams may or may not finish.

Installation

yarn add @x/observable

Usage

Creating Observable Types

const observableInstance = observable((publish, observable) => {})

Creation of an observable object is done using the exported observable factory function. The function accepts a single parameter, a definition function that describes how and when new values are emitted. The definition function accepts two parameters - the publish function and the observable being constructed. Functions returned from the function is attached to the observable as the disconnect function.

const { observable } = require('@x/observable')
  
// a simple observable that emits a count at a defined interval
const timerObservable = (interval = 1000) => observable((publish, instance) => {
  let count = 0
  
  // set a timer to publish the next count at the specified interval
  const handle = setInterval(
    () => publish(++count), 
    interval
  )
  
  // provide a disconnect function to allow consumers to stop the timer
  return () => clearTimeout(handle)
})

Consuming Observables

const subscription = observableInstance.subscribe(handler, onError)

Listening for new values emitted by observables is done by calling the subscribe function, passing a callback to be executed for each new value. The subscribe function returns a subscription object that can be used to unsubscribe from further values.

// create a new instance of the timer observable type we created above
const timer = timerObservable(500)

// execute a function every time the observable emits a new value
const subscription = timer.subscribe(
  count => console.log(`Observable has emitted ${count} values`)
)

// unsubscribe from the observable after five seconds
setTimeout(
  () => {
    subscription.unsubscribe()
    console.log(`We observed ${timer()} values being emitted`)
  },
  5000
)

Handling Errors

The subscribe function accepts a second parameter named onError that should be passed a callback that is called when an error occurs in the subscription handler.

Built In Observable Factories

Two built in observable types are provided, subject and proxy.

subject

The subject observable type is a simple observable that allows external consumers to publish new events through the observable through the exposed publish function:

const { subject } = require('@x/observable')

const source = subject()
source.publish({ value: 10 })
console.log(source())

proxy

The proxy observable type can be used to wrap another observable, allowing consumers to disconnect from the source using a disconnect function, without the returned subscription object, and without impacting the source observable in any way:

const { proxy, subject } = require('@x/observable')

const source = subject()
const proxied = proxy(source)
subject.publish({ value: 20 })
proxied.disconnect()
subject.publish({ value: 30 })

Consuming Event Emitter Objects

A fromEmitter factory is provided to allow simple subscription to events emitted by traditional EventEmitter objects. Both Node.js and browser style objects are supported. Multiple events can be specified on the fromEmitter function call. The topic property of emitted events contains the original event name, and the data property contains the event data. Other arguments passed to event handlers can be accessed using the args property.

const { fromEmitter } = require('@x/observable')

// consume a browser style event emitter
const resizeObservable = fromEmitter(window, 'resize')
resizeObservable.subscribe(() => {
  console.log(`Width: ${window.innerWidth}, height: ${window.innerHeight}`)
})

// consume a Node.js style event emitter
const messageObservable = fromEmitter(process, 'message')
messageObservable.subscribe(({ topic, data }) => {
  console.log(`Event topic was ${topic}. Event data was ${JSON.stringify(data)}`)
})

Merging Multiple Observables

Three helper functions are provided that allow merging multiple observables into a single observable:

merge

The merge function creates a new observable from provided source observables and emits a new value every time one of the sources emits a new value.

const { merge, subject } = require('@x/observable')

const source1 = subject()
const source2 = subject()
const merged = merge(source1, source2)
merged.subscribe(value => console.log(value))
source1.publish(1) // logs `1`
source2.publish('test') // logs `test`

mergeHash

The mergeHash function creates a new observable from a hash of multiple observables.

const { mergeHash, subject } = require('@x/observable')

const source1 = subject()
const source2 = subject()
const merged = merge({ source1, source2 })
merged.subscribe(value => console.log(JSON.stringify(value)))
source1.publish(1) // logs `{"source1":1,"source2":undefined}`
source2.publish('test') // logs `{"source1":1,"source2":"test"}`

mergeArray

The mergeArray function creates a new observable from multiple observables.

const { mergeArray, subject } = require('@x/observable')

const source1 = subject()
const source2 = subject()
const merged = mergeArray(source1, source2)
merged.subscribe(value => console.log(JSON.stringify(value)))
source1.publish(1) // logs `[1,undefined]`
source2.publish('test') // logs `[1,"test"]`

Hot Swapping Source Observables

The swappable observable type allows dynamic replacement of source observables.

const { swappable, subject } = require('@x/observable')

const source1 = subject()
const swappableObservable = swappable(source1, { publishOnSwap: true })
const source2 = subject()
swappableObservable.swap(source2)

Utility Functions

Several utility functions are provided for common scenarios.

isObservable

The isObservable function allows simple testing to see if an object is an observable object.

const { isObservable, subject } = require('@x/observable')

const source = subject()
console.log(isObservable(source)) // logs 'true'
console.log(isObservable(window)) // logs 'false'

unwrap

The unwrap function recursively evaluates all observables exposed on an object.

const { unwrap, subject } = require('@x/observable')

const source1 = subject({ initialValue: 1 })
const source2 = subject({ initialValue: 'test' })
console.log(unwrap(source1)) // logs `1`
console.log(JSON.stringify(unwrap({ source1, source2 }))) // logs `{"source1":1,"source2":"test"}`

Observable Options

The second parameter to observable factory functions is an object containing any number of options, as described below:

Name|Type|Description ----|----|----------- initialValue|any|The initial value encapsulated by the observable

License

The MIT License (MIT)

Copyright © 2022 Dale Anderson

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.