stream-patterns

Stream partitioning utility that splits a stream into multiple streams based on pattern matching.

Installation

npm install stream-patterns

Usage

const { partition } = require('stream-patterns')
const { Readable } = require('streamx')

// Create a source stream with mixed data
const source = Readable.from([
  { type: 'user', name: 'Alice' },
  { type: 'system', msg: 'Starting' },
  { type: 'user', name: 'Bob' },
  { type: 'system', msg: 'Complete' }
])

// Partition the stream by patterns
const patterns = partition(source, { type: 'user' }, { type: 'system' })

// Get data for each pattern
const users = await patterns.get({ type: 'user' })
const system = await patterns.get({ type: 'system' })

console.log(users)
// [{ type: 'user', name: 'Alice' }, { type: 'user', name: 'Bob' }]

console.log(system)
// [{ type: 'system', msg: 'Starting' }, { type: 'system', msg: 'Complete' }]

API

partition(stream, ...patterns[, done])

Partitions a stream into multiple streams based on pattern matching.

• stream - Source stream to partition
• ...patterns - Pattern objects to match against stream data (uses ptnm for matching)
• done - Optional callback function (err) => {} called when partitioning is complete

Returns a Map where keys are the pattern objects and values are Promises that resolve to arrays of matched data.

Example with callback:

const patterns = partition(stream, { type: 'error' }, { type: 'info' }, (err) => {
  if (err) console.error('Partition error:', err)
  else console.log('Partitioning complete')
})

Pattern

A writable stream class that handles pattern matching and routing.

const { Pattern } = require('stream-patterns')
const { Readable, pipeline } = require('streamx')

const pattern = new Pattern({ type: 'a' }, { type: 'b' })
const source = Readable.from([{ type: 'a' }, { type: 'b' }])

pipeline(source, pattern, (err) => {
  if (err) console.error(err)
})

// Access matched streams
const streamA = pattern.streams[0]
const streamB = pattern.streams[1]

Error Handling

Errors from the source stream propagate to both the partition callback and the individual pattern promises:

const patterns = partition(errorStream, { type: 'data' }, (err) => {
  // Error caught here
})

const data = patterns.get({ type: 'data' })
data.catch((err) => {
  // Also caught here
})

License

MIT