Documentation Index
Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
Use this file to discover all available pages before exploring further.
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>
Show StreamGetInput properties
The item data, or null if not found
set()
Set an item in the stream.
set(input: StreamSetInput): Promise<StreamSetResult<TData> | null>
Show StreamSetInput properties
result
StreamSetResult<TData> | null
Result containing old and new valuesShow StreamSetResult properties
delete()
Delete an item from the stream.
delete(input: StreamDeleteInput): Promise<DeleteResult>
input
StreamDeleteInput
required
Show StreamDeleteInput properties
Show DeleteResult properties
The deleted value (if any)
list()
List all items in a group.
list(input: StreamListInput): Promise<TData[]>
Show StreamListInput properties
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
update()
Perform atomic updates on an item.
update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData> | null>
input
StreamUpdateInput
required
Show StreamUpdateInput properties
Array of update operations
Set a value:{ type: 'set', path: 'field.nested', value: any }
{ type: 'increment', path: 'counter', by: 1 }
{ type: 'decrement', path: 'counter', by: 1 }
{ type: 'remove', path: 'field' }
{ type: 'merge', path: 'object', value: { key: 'value' } }
result
StreamUpdateResult<TData> | null
Show StreamUpdateResult properties
createStream()
Register a stream implementation with the III SDK.
iii.createStream<TData>(streamName, stream)
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
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 })
}
}
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 }
}
}
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
}