Streaming

Create custom streaming data sources with the IStream interface

The IStream interface allows you to create custom streaming data sources for real-time collaborative data, pub/sub systems, or stateful streams.

IStream Interface

Implement the IStream<TData> interface to create custom stream backends.

interface IStream<TData> {
  get(input: StreamGetInput): Promise<TData | null>
  set(input: StreamSetInput): Promise<StreamSetResult<TData> | null>
  delete(input: StreamDeleteInput): Promise<DeleteResult>
  list(input: StreamListInput): Promise<TData[]>
  listGroups(input: StreamListGroupsInput): Promise<string[]>
  update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData> | null>
}

Stream Operations

get()

Retrieve an item from the stream.

get(input: StreamGetInput): Promise<TData | null>

input · StreamGetInput · required

Show StreamGetInput properties

stream_name · string · required

Name of the stream

group_id · string · required

Group/namespace ID

item_id · string · required

Unique item ID

data · TData | null

The item data, or null if not found

set()

Set an item in the stream.

set(input: StreamSetInput): Promise<StreamSetResult<TData> | null>

input · StreamSetInput · required

Show StreamSetInput properties

stream_name · string · required

Name of the stream

group_id · string · required

Group/namespace ID

item_id · string · required

Unique item ID

data · any · required

Data to set

result · StreamSetResult<TData> | null

Result containing old and new values

Show StreamSetResult properties

old_value · TData

Previous value (if any)

new_value · TData · required

New value that was set

delete()

Delete an item from the stream.

delete(input: StreamDeleteInput): Promise<DeleteResult>

input · StreamDeleteInput · required

Show StreamDeleteInput properties

stream_name · string · required

Name of the stream

group_id · string · required

Group/namespace ID

item_id · string · required

Unique item ID

result · DeleteResult

Show DeleteResult properties

old_value · any

The deleted value (if any)

list()

List all items in a group.

list(input: StreamListInput): Promise<TData[]>

input · StreamListInput · required

Show StreamListInput properties

stream_name · string · required

Name of the stream

group_id · string · required

Group/namespace ID

items · TData[]

Array of all items in the group

listGroups()

List all groups in the stream.

listGroups(input: StreamListGroupsInput): Promise<string[]>

input · StreamListGroupsInput · required

Show StreamListGroupsInput properties

stream_name · string · required

Name of the stream

groups · string[]

Array of group IDs

update()

Perform atomic updates on an item.

update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData> | null>

input · StreamUpdateInput · required

Show StreamUpdateInput properties

stream_name · string · required

Name of the stream

group_id · string · required

Group/namespace ID

item_id · string · required

Unique item ID

ops · UpdateOp[] · required

Array of update operations

Show Update operations

Set a value:

{ type: 'set', path: 'field.nested', value: any }

Increment a number:

{ type: 'increment', path: 'counter', by: 1 }

Decrement a number:

{ type: 'decrement', path: 'counter', by: 1 }

Remove a field:

{ type: 'remove', path: 'field' }

Merge an object:

{ type: 'merge', path: 'object', value: { key: 'value' } }

result · StreamUpdateResult<TData> | null

Show StreamUpdateResult properties

old_value · TData

Value before update

new_value · TData · required

Value after update

createStream()

Register a stream implementation with the III SDK.

iii.createStream<TData>(streamName, stream)

streamName · string · required

Unique stream name

stream · IStream<TData> · required

Stream implementation

This automatically registers the following functions:

• stream::get(streamName)
• stream::set(streamName)
• stream::delete(streamName)
• stream::list(streamName)
• stream::list_groups(streamName)

Example: In-Memory Stream

import { init, type IStream, type StreamGetInput, type StreamSetInput, type StreamDeleteInput, type StreamListInput, type StreamListGroupsInput } from 'iii-sdk'

const iii = init('ws://localhost:49199')

interface TodoItem {
  id: string
  title: string
  completed: boolean
  created_at: string
}

class InMemoryStream implements IStream<TodoItem> {
  private data = new Map<string, Map<string, TodoItem>>()
  
  async get(input: StreamGetInput): Promise<TodoItem | null> {
    const group = this.data.get(input.group_id)
    return group?.get(input.item_id) ?? null
  }
  
  async set(input: StreamSetInput) {
    let group = this.data.get(input.group_id)
    if (!group) {
      group = new Map()
      this.data.set(input.group_id, group)
    }
    
    const old_value = group.get(input.item_id)
    group.set(input.item_id, input.data)
    
    return {
      old_value,
      new_value: input.data
    }
  }
  
  async delete(input: StreamDeleteInput) {
    const group = this.data.get(input.group_id)
    const old_value = group?.get(input.item_id)
    group?.delete(input.item_id)
    
    return { old_value }
  }
  
  async list(input: StreamListInput): Promise<TodoItem[]> {
    const group = this.data.get(input.group_id)
    return group ? Array.from(group.values()) : []
  }
  
  async listGroups(input: StreamListGroupsInput): Promise<string[]> {
    return Array.from(this.data.keys())
  }
  
  async update(input: any) {
    // Implement atomic updates if needed
    return null
  }
}

// Register the stream
iii.createStream('todos', new InMemoryStream())

// Now you can access it via functions
const setResult = await iii.call('stream::set(todos)', {
  stream_name: 'todos',
  group_id: 'user-123',
  item_id: 'todo-1',
  data: {
    id: 'todo-1',
    title: 'Buy milk',
    completed: false,
    created_at: new Date().toISOString()
  }
})

const todo = await iii.call('stream::get(todos)', {
  stream_name: 'todos',
  group_id: 'user-123',
  item_id: 'todo-1'
})

const allTodos = await iii.call('stream::list(todos)', {
  stream_name: 'todos',
  group_id: 'user-123'
})

Example: Redis Stream

import { init, type IStream } from 'iii-sdk'
import { createClient } from 'redis'

const iii = init('ws://localhost:49199')
const redis = createClient()
await redis.connect()

class RedisStream<TData> implements IStream<TData> {
  constructor(private redis: ReturnType<typeof createClient>) {}
  
  private getKey(group_id: string, item_id: string): string {
    return `stream:${group_id}:${item_id}`
  }
  
  async get(input: any): Promise<TData | null> {
    const key = this.getKey(input.group_id, input.item_id)
    const data = await this.redis.get(key)
    return data ? JSON.parse(data) : null
  }
  
  async set(input: any) {
    const key = this.getKey(input.group_id, input.item_id)
    const old_value = await this.get(input)
    
    await this.redis.set(key, JSON.stringify(input.data))
    
    return {
      old_value,
      new_value: input.data
    }
  }
  
  async delete(input: any) {
    const key = this.getKey(input.group_id, input.item_id)
    const old_value = await this.get(input)
    
    await this.redis.del(key)
    
    return { old_value }
  }
  
  async list(input: any): Promise<TData[]> {
    const pattern = `stream:${input.group_id}:*`
    const keys = await this.redis.keys(pattern)
    
    const items: TData[] = []
    for (const key of keys) {
      const data = await this.redis.get(key)
      if (data) {
        items.push(JSON.parse(data))
      }
    }
    
    return items
  }
  
  async listGroups(input: any): Promise<string[]> {
    const pattern = 'stream:*'
    const keys = await this.redis.keys(pattern)
    
    const groups = new Set<string>()
    for (const key of keys) {
      const parts = key.split(':')
      if (parts.length >= 2) {
        groups.add(parts[1])
      }
    }
    
    return Array.from(groups)
  }
  
  async update(input: any) {
    // Implement using Redis transactions
    return null
  }
}

iii.createStream('sessions', new RedisStream(redis))

Example: Database Stream

import { init, type IStream } from 'iii-sdk'
import { PrismaClient } from '@prisma/client'

const iii = init('ws://localhost:49199')
const prisma = new PrismaClient()

interface Document {
  id: string
  title: string
  content: string
  updated_at: string
}

class DatabaseStream implements IStream<Document> {
  async get(input: any): Promise<Document | null> {
    const doc = await prisma.document.findFirst({
      where: {
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return doc ? {
      id: doc.id,
      title: doc.title,
      content: doc.content,
      updated_at: doc.updated_at.toISOString()
    } : null
  }
  
  async set(input: any) {
    const old_value = await this.get(input)
    
    const doc = await prisma.document.upsert({
      where: {
        group_id_id: {
          group_id: input.group_id,
          id: input.item_id
        }
      },
      update: input.data,
      create: {
        ...input.data,
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return {
      old_value,
      new_value: {
        id: doc.id,
        title: doc.title,
        content: doc.content,
        updated_at: doc.updated_at.toISOString()
      }
    }
  }
  
  async delete(input: any) {
    const old_value = await this.get(input)
    
    await prisma.document.deleteMany({
      where: {
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return { old_value }
  }
  
  async list(input: any): Promise<Document[]> {
    const docs = await prisma.document.findMany({
      where: { group_id: input.group_id }
    })
    
    return docs.map(doc => ({
      id: doc.id,
      title: doc.title,
      content: doc.content,
      updated_at: doc.updated_at.toISOString()
    }))
  }
  
  async listGroups(): Promise<string[]> {
    const groups = await prisma.document.findMany({
      select: { group_id: true },
      distinct: ['group_id']
    })
    
    return groups.map(g => g.group_id)
  }
  
  async update(input: any) {
    // Implement using Prisma transactions
    return null
  }
}

iii.createStream('documents', new DatabaseStream())

Use Cases

Real-time Collaboration

Implement operational transformation or CRDTs for collaborative editing:

class CollaborativeDocStream implements IStream<Document> {
  // Implement get, set, delete, list, listGroups
  
  async update(input: any) {
    // Apply operational transforms
    const doc = await this.get(input)
    const updated = applyOperations(doc, input.ops)
    return await this.set({ ...input, data: updated })
  }
}

Event Sourcing

Store events and compute state from event history:

class EventSourcedStream implements IStream<State> {
  async set(input: any) {
    // Append event
    await appendEvent(input.data)
    
    // Recompute state
    const events = await getEvents(input.group_id)
    const new_value = computeState(events)
    
    return { old_value: null, new_value }
  }
}

Caching Layer

Add caching on top of another data source:

class CachedStream implements IStream<Data> {
  constructor(
    private backend: IStream<Data>,
    private cache: Map<string, Data>
  ) {}
  
  async get(input: any) {
    const key = `${input.group_id}:${input.item_id}`
    
    if (this.cache.has(key)) {
      return this.cache.get(key)!
    }
    
    const data = await this.backend.get(input)
    if (data) this.cache.set(key, data)
    
    return data
  }
  
  // Implement other methods with cache invalidation
}