Server-side Middleware

def my_middleware(patch: StreamingChunk, next_fn: Callable) -> None:
    # transform, drop, or fan out
    next_fn(patch)
 
emitter.use(my_middleware)

emitter.use((patch, next) => {
  // transform, drop, or fan out
  next(patch)
})

def resolve_images(patch, next_fn):
    if patch.op == 'add' and isinstance(patch.value, str):
        if patch.value.startswith('{{img:'):
            filename = patch.value[6:-2]
            patch = patch.replace(value=get_presigned_url(filename))
    next_fn(patch)
 
emitter.use(resolve_images)

emitter.use((patch, next) => {
  if (patch.op === 'add' && typeof patch.value === 'string') {
    if (patch.value.startsWith('{{img:')) {
      const filename = patch.value.slice(6, -2)
      return next({ ...patch, value: getPresignedUrl(filename) })
    }
  }
  next(patch)
})

def strip_internal(patch, next_fn):
    if 'internal' not in patch.path:
        next_fn(patch)
 
emitter.use(strip_internal)

emitter.use((patch, next) => {
  if (!patch.path.includes('internal')) next(patch)
})

def inject_author(patch, next_fn):
    next_fn(patch)
    if patch.path == 'authorId' and patch.op == 'complete':
        author = db.get_user(patch.value)
        next_fn(StreamingChunk(path='author', value=author, op='add'))
 
emitter.use(inject_author)

emitter.use(async (patch, next) => {
  next(patch)
  if (patch.path === 'authorId' && patch.op === 'complete') {
    const author = await db.getUser(patch.value as string)
    next({ path: 'author', value: author, op: 'add' })
  }
})