Streaming

Kinotic services can push multiple values to callers over time using the streaming pattern. Instead of returning a single response, a streaming method returns an Observable that emits values as they become available.

Defining a Streaming Method

Any service method that returns an Observable is treated as a streaming endpoint.

import { Publish } from '@kinotic-ai/core'
import { Observable, interval } from 'rxjs'
import { map } from 'rxjs/operators'

@Publish('com.example')
class SensorService {
    streamReadings(sensorId: string): Observable<SensorReading> {
        return interval(1000).pipe(
            map(() => ({
                sensorId,
                value: Math.random() * 100,
                timestamp: new Date().toISOString()
            }))
        )
    }
}

Consuming a Stream

Callers receive an Observable and subscribe to process values as they arrive.

const stream$ = sensorService.streamReadings('sensor-1')

const subscription = stream$.subscribe({
    next: (reading) => console.log('Reading:', reading.value),
    error: (err) => console.error('Stream error:', err),
    complete: () => console.log('Stream ended')
})

// Stop listening when done
subscription.unsubscribe()

How a Stream Ends

A stream ends for its caller in one of four ways, and the service stops producing in every one of them:

• The service completes it. The caller's Observable completes.
• The service fails it. The caller's Observable errors with the exception the service threw or emitted.
• The caller unsubscribes. A cancel reaches the service, which unsubscribes from the Observable the method returned, so a finite source stops early and an interval stops ticking.
• The caller is gone. The platform watches the caller's reply destination for every stream in progress, on the node hosting a Java service and on the gateway for a TypeScript one. When nothing listens there any more — the caller's connection closed, or its node left the cluster — the stream is cancelled the same way an unsubscribe cancels it. A TypeScript service whose own connection ends, however it ends, cancels every stream it was producing at that moment, because the gateway has already failed those callers (see service call liveness).

A service that stops while producing a stream ends it with an error at the caller rather than a silent cancel.

Use Cases

• Real-time data feeds -- IoT sensors, live metrics, dashboards
• Notifications and alerts -- Push updates to connected clients
• Change data capture -- React to data changes as they happen
• Large result sets -- Stream results incrementally instead of loading everything at once