🟢 Beta SDK — This SDK is feature-complete with tests and CI for iOS/macOS. For browser/WebAssembly use cases, also see the WASM SDK. Contributions welcome!

Streamline Swift SDK

Swift client SDK for Streamline — The Redis of Streaming.

Requirements

• Swift 5.9+
• iOS 15+ / macOS 13+
• Streamline server 0.2.0 or later

Installation

Swift Package Manager

Add the dependency to your Package.swift:

dependencies: [
    .package(url: "https://github.com/streamlinelabs/streamline-swift-sdk.git", from: "0.2.0"),
]

Quick Start

import StreamlineSDK

let config = StreamlineConfiguration(
    url: URL(string: "ws://localhost:9092")!,
    authToken: "my-token"
)

let client = StreamlineClient(configuration: config)
client.connect()

// Produce a message
try client.produce(topic: "events", key: "user-1", stringValue: "{\"action\":\"click\"}")

// Subscribe to a topic
client.subscribe(topic: "events") { message in
    print("Received: \(String(data: message.value, encoding: .utf8) ?? "")")
}

// Disconnect when done
client.disconnect()

Transactions

let client = StreamlineClient(configuration: config)
try await client.connect()

try await client.beginTransaction()
do {
    try await client.produce(topic: "orders", key: "k1", value: "v1")
    try await client.produce(topic: "orders", key: "k2", value: "v2")
    try await client.commitTransaction()
} catch {
    try await client.abortTransaction()
    throw error
}

Note: Transactions use client-side buffering. Messages are collected and sent as a batch on commit, providing all-or-nothing delivery at the client level.

Admin Client

The AdminClient communicates with the Streamline HTTP REST API (port 9094) for topic management, consumer group inspection, and SQL queries.

let admin = AdminClient(baseURL: URL(string: "http://localhost:9094")!, authToken: "my-token")

// Topic management
try await admin.createTopic(name: "events", partitions: 3)
let topics = try await admin.listTopics()
let details = try await admin.describeTopic(name: "events")
try await admin.deleteTopic(name: "old-topic")

// Consumer groups
let groups = try await admin.listConsumerGroups()
let groupDetails = try await admin.describeConsumerGroup(groupId: "my-group")

// SQL queries
let result = try await admin.query("SELECT * FROM events LIMIT 10")
for row in result.rows { print(row) }

// Server info
let info = try await admin.serverInfo()
print("Version: \(info.version), Topics: \(info.topicCount)")

Cluster & Monitoring

let admin = AdminClient(baseURL: URL(string: "http://localhost:9094")!)

// Cluster overview
let cluster = try await admin.clusterInfo()
print("Cluster: \(cluster.clusterId), Brokers: \(cluster.brokers.count)")

// Consumer group lag monitoring
let lag = try await admin.consumerGroupLag(groupId: "my-group")
print("Total lag: \(lag.totalLag)")
for p in lag.partitions { print("  \(p.topic):\(p.partition) lag=\(p.lag)") }

// Message inspection
let messages = try await admin.inspectMessages(topic: "events", partition: 0, limit: 10)
for m in messages { print("offset=\(m.offset) value=\(m.value)") }

// Latest messages
let latest = try await admin.latestMessages(topic: "events", count: 5)

// Server metrics
let metrics = try await admin.metricsHistory()
for m in metrics { print("\(m.name)=\(m.value)") }

// Offset management
let dryRun = try await admin.resetOffsetsDryRun(groupId: "my-group", topic: "events")
try await admin.resetOffsets(groupId: "my-group", topic: "events")

AsyncStream Consumption

Consume messages using Swift's structured concurrency with for await:

// Stream messages as an AsyncStream
for await message in client.messages(topic: "events") {
    let value = String(data: message.value, encoding: .utf8) ?? ""
    print("Key: \(message.key ?? "nil"), Value: \(value)")
}

Schema Registry

The SDK includes a full Schema Registry client:

let registry = SchemaRegistryClient(baseURL: URL(string: "http://localhost:9094")!)

// Register a schema
let id = try await registry.registerSchema(
    subject: "events-value", schema: avroJson, format: .avro
)

// Retrieve the latest schema
let schema = try await registry.getLatestSchema(subject: "events-value")
print("Version: \(schema.version), Type: \(schema.schemaType)")

// Check compatibility
let compatible = try await registry.checkCompatibility(
    subject: "events-value", schema: newSchema, format: .avro
)

// List subjects and versions
let subjects = try await registry.listSubjects()
let versions = try await registry.listVersions(subject: "events-value")

Supports AVRO, PROTOBUF, and JSON schema formats.

Security

TLS

let config = StreamlineConfiguration(
    url: URL(string: "wss://streamline.example.com:9092")!,
    tls: TlsConfig(enabled: true, caCertificatePath: "/etc/ssl/ca.pem")
)

SASL Authentication

let config = StreamlineConfiguration(
    url: URL(string: "ws://streamline.example.com:9092")!,
    sasl: SaslConfig(mechanism: .scramSha256, username: "admin", password: "secret")
)

Producer & Consumer Configuration

let config = StreamlineConfiguration(
    url: URL(string: "ws://localhost:9092")!,
    producerConfig: ProducerConfig(batchSize: 32768, compression: .lz4, acks: .all),
    consumerConfig: ConsumerConfig(groupId: "my-app", autoCommit: false, autoOffsetReset: .earliest)
)

Features

• WebSocket connection to Streamline server
• Admin client — topic CRUD, consumer groups, SQL queries via HTTP REST API
• Schema Registry — register, retrieve, and validate schemas (Avro, Protobuf, JSON)
• Security — TLS encryption and SASL authentication (PLAIN, SCRAM-SHA-256/512)
• Producer/Consumer config — batching, compression, acknowledgments, consumer groups
• AsyncStream consumption — idiomatic for await streaming with automatic lifecycle
• Telemetry — pluggable tracing with W3C Trace Context propagation
• Auto-reconnect with exponential backoff
• Offline message queue — messages produced while disconnected are buffered and sent on reconnect
• Delegate pattern for connection lifecycle events
• Sendable-safe types for structured concurrency

Configuration

Parameter	Default	Description
url	(required)	WebSocket URL of the Streamline server
autoReconnect	true	Automatically reconnect on disconnection
maxRetries	10	Maximum reconnection attempts
timeout	30	Connection timeout in seconds
authToken	nil	Optional bearer token for authentication
tls	nil	TLS configuration (see Security)
sasl	nil	SASL authentication (see Security)
producerConfig	defaults	Producer tuning (see Producer & Consumer Configuration)
consumerConfig	defaults	Consumer tuning (see Producer & Consumer Configuration)
initialBackoff	0.5	Initial reconnection backoff (seconds)
maxBackoff	30	Maximum backoff cap (seconds)

Error Handling

All SDK errors are represented by the StreamlineError enum. Each case provides context about the failure:

Error	Description	Retryable?
.notConnected	Client is not connected to the server	Yes — reconnects automatically
.connectionFailed(String)	Connection attempt failed with reason	Yes — retry with backoff
.authenticationFailed(String)	Server rejected credentials	No
.timeout	Operation timed out	Yes
.topicNotFound(String)	Requested topic does not exist	No — create the topic first
.serializationError(String)	Message encoding/decoding failed	No
.offlineQueueFull	Offline buffer capacity exceeded	No — reduce send rate
.adminOperationFailed(String)	Admin API call failed	Depends on cause
.queryFailed(String)	SQL query execution failed	Depends on cause
.schemaRegistryError(String)	Schema registry operation failed	Depends on cause

do {
    try client.produce(topic: "my-topic", key: "key", value: Data("value".utf8))
} catch let error as StreamlineError {
    switch error {
    case .notConnected:
        print("Not connected — messages are queued offline")
    case .connectionFailed(let reason):
        print("Connection failed: \(reason)")
    case .topicNotFound(let name):
        print("Topic not found: \(name)")
    case .timeout:
        print("Operation timed out — consider increasing timeout")
    case .offlineQueueFull:
        print("Offline queue full — reduce send rate or increase queue size")
    default:
        print("Error: \(error.localizedDescription)")
    }
}

Retry Strategy

The Swift SDK automatically retries failed sends with exponential backoff when ProducerConfig.retries > 0 (default: 3). Configure retry behavior:

let config = ProducerConfig(
    retries: 5,              // Max retry attempts
    retryBackoffMs: 200      // Base backoff (doubles each attempt)
)
let client = StreamlineClient(
    configuration: StreamlineConfiguration(url: wsURL),
    producerConfig: config
)

Circuit Breaker

Protect your application from cascading failures when the Streamline server is unresponsive:

import StreamlineSDK

let breaker = CircuitBreaker(config: CircuitBreakerConfig(
    failureThreshold: 5,      // Open after 5 consecutive failures
    successThreshold: 2,      // Close after 2 half-open successes
    openTimeout: 30.0         // 30s before probing
))

if breaker.check() {
    do {
        try client.produce(topic: "events", key: "user-1", stringValue: "payload")
        breaker.recordSuccess()
    } catch {
        breaker.recordFailure()
        throw error
    }
}

When the circuit is open, check() returns false and operations should be skipped. See the Circuit Breaker guide for details.

Examples

The examples/ directory contains runnable examples:

Example	Description
BasicUsage.swift	Produce, consume, and admin operations
QueryUsage.swift	SQL analytics with the embedded query engine
SchemaRegistryUsage.swift	Schema registration and validation
CircuitBreakerUsage.swift	Resilient production with circuit breaker
SecurityUsage.swift	TLS and SASL authentication

Moonshot Features

⚠️ Experimental — These features require Streamline server 0.3.0+ with moonshot feature flags enabled.

Semantic Search

Query topics by meaning instead of offset. Requires a topic created with semantic.embed=true.

let results = try await client.search(topic: "logs.app", query: "payment failure", k: 10)
for hit in results {
    print("[p\(hit.partition)] offset=\(hit.offset) score=\(String(format: "%.2f", hit.score))")
}

Attestation Verification

Verify cryptographic provenance attestations attached to records by data contracts.

import StreamlineSDK

let verifier = try StreamlineVerifier(publicKey: publicKeyData)
let result = try verifier.verify(record: record)
print("Verified: \(result.verified), Producer: \(result.producerId)")

Agent Memory (MCP)

Use Streamline as persistent memory for AI agents via the MCP protocol.

let memory = MemoryClient(baseURL: URL(string: "http://localhost:9094/mcp/v1")!)
try await memory.remember("user prefers dark mode", tags: ["preferences"])
let results = try await memory.recall("user preferences", k: 5)

Branched Streams

Create topic branches for replay, A/B testing, or counterfactual analysis.

let branch = try await admin.createBranch(topic: "events", name: "experiment-v2")
for await msg in client.messages(topic: branch.topic) {
    process(msg)
}

Contributing

Contributions are welcome! This is a community-maintained SDK. Please see the organization contributing guide for guidelines.

License

Apache-2.0