IntercomServer binds web server (8080) and media port (50000) to the clinic interface only. Phones post an SDP offer and receive talkback (Opus) and the PGM preview (H.264 constrained baseline, 960x540, 25 fps). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>main
parent
7d885ecd57
commit
7801d04b44
@ -0,0 +1,155 @@
|
||||
import Foundation
|
||||
import Network
|
||||
import OSLog
|
||||
|
||||
struct HTTPRequest: Equatable {
|
||||
var method: String
|
||||
var path: String
|
||||
var headers: [String: String]
|
||||
var body: [UInt8]
|
||||
}
|
||||
|
||||
struct HTTPResponse {
|
||||
var status: Int
|
||||
var contentType: String
|
||||
var body: [UInt8]
|
||||
|
||||
static func text(_ status: Int, _ text: String) -> HTTPResponse {
|
||||
HTTPResponse(status: status, contentType: "text/plain; charset=utf-8", body: Array(text.utf8))
|
||||
}
|
||||
|
||||
static func json(_ value: some Encodable) -> HTTPResponse {
|
||||
HTTPResponse(status: 200, contentType: "application/json", body: Array((try? JSONEncoder().encode(value)) ?? Data()))
|
||||
}
|
||||
|
||||
var serialized: [UInt8] {
|
||||
let reason = [200: "OK", 204: "No Content", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed",
|
||||
413: "Payload Too Large", 503: "Service Unavailable"][status] ?? "Error"
|
||||
let head = "HTTP/1.1 \(status) \(reason)\r\nContent-Type: \(contentType)\r\nContent-Length: \(body.count)\r\n"
|
||||
+ "Cache-Control: no-store\r\nConnection: close\r\n\r\n"
|
||||
return Array(head.utf8) + body
|
||||
}
|
||||
}
|
||||
|
||||
/// Incremental HTTP/1.1 request parser (no chunked bodies; the phone page only sends small JSON).
|
||||
struct HTTPRequestParser {
|
||||
enum ParseError: Error { case malformed, tooLarge }
|
||||
|
||||
static let maxHeader = 8 * 1024
|
||||
static let maxBody = 64 * 1024
|
||||
|
||||
private var buffer: [UInt8] = []
|
||||
|
||||
/// Appends bytes; returns the request once it is complete.
|
||||
mutating func append(_ bytes: [UInt8]) throws -> HTTPRequest? {
|
||||
buffer += bytes
|
||||
guard let headerEnd = Self.find([13, 10, 13, 10], in: buffer) else {
|
||||
if buffer.count > Self.maxHeader { throw ParseError.tooLarge }
|
||||
return nil
|
||||
}
|
||||
guard let head = String(bytes: buffer[0..<headerEnd], encoding: .utf8) else { throw ParseError.malformed }
|
||||
let lines = head.components(separatedBy: "\r\n")
|
||||
let requestLine = lines[0].split(separator: " ")
|
||||
guard requestLine.count == 3, requestLine[2].hasPrefix("HTTP/1.") else { throw ParseError.malformed }
|
||||
var headers: [String: String] = [:]
|
||||
for line in lines.dropFirst() {
|
||||
guard let colon = line.firstIndex(of: ":") else { continue }
|
||||
headers[line[..<colon].lowercased()] = line[line.index(after: colon)...].trimmingCharacters(in: .whitespaces)
|
||||
}
|
||||
let length = Int(headers["content-length"] ?? "0") ?? -1
|
||||
guard length >= 0 else { throw ParseError.malformed }
|
||||
guard length <= Self.maxBody else { throw ParseError.tooLarge }
|
||||
let bodyStart = headerEnd + 4
|
||||
guard buffer.count >= bodyStart + length else { return nil }
|
||||
let path = String(requestLine[1].split(separator: "?", maxSplits: 1).first ?? "/")
|
||||
return HTTPRequest(method: String(requestLine[0]), path: path, headers: headers,
|
||||
body: Array(buffer[bodyStart..<bodyStart + length]))
|
||||
}
|
||||
|
||||
private static func find(_ needle: [UInt8], in haystack: [UInt8]) -> Int? {
|
||||
guard haystack.count >= needle.count else { return nil }
|
||||
for i in 0...(haystack.count - needle.count) where haystack[i] == needle[0] {
|
||||
if Array(haystack[i..<i + needle.count]) == needle { return i }
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
/// Minimal HTTP server bound to one interface. One request per connection.
|
||||
final class HTTPServer: @unchecked Sendable {
|
||||
typealias Handler = @Sendable (HTTPRequest) -> HTTPResponse
|
||||
|
||||
private let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.http")
|
||||
private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")
|
||||
private var listener: NWListener?
|
||||
private let handler: Handler
|
||||
|
||||
init(handler: @escaping Handler) {
|
||||
self.handler = handler
|
||||
}
|
||||
|
||||
func start(interface: NWInterface, port: UInt16) async throws {
|
||||
let parameters = NWParameters.tcp
|
||||
parameters.requiredInterface = interface
|
||||
parameters.allowLocalEndpointReuse = true
|
||||
parameters.includePeerToPeer = false
|
||||
let listener = try NWListener(using: parameters, on: NWEndpoint.Port(rawValue: port) ?? .any)
|
||||
let once = OnceFlag()
|
||||
try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
|
||||
listener.stateUpdateHandler = { [log] state in
|
||||
switch state {
|
||||
case .ready:
|
||||
if once.claim() { continuation.resume() }
|
||||
case .failed(let error):
|
||||
log.error("HTTP failed: \(error.localizedDescription, privacy: .public)")
|
||||
if once.claim() { continuation.resume(throwing: error) }
|
||||
default: break
|
||||
}
|
||||
}
|
||||
listener.newConnectionHandler = { [weak self] connection in self?.serve(connection) }
|
||||
listener.start(queue: queue)
|
||||
}
|
||||
self.listener = listener
|
||||
}
|
||||
|
||||
func stop() {
|
||||
listener?.cancel()
|
||||
listener = nil
|
||||
}
|
||||
|
||||
private func serve(_ connection: NWConnection) {
|
||||
connection.start(queue: queue)
|
||||
read(connection, parser: ParserBox())
|
||||
}
|
||||
|
||||
private func read(_ connection: NWConnection, parser: ParserBox) {
|
||||
connection.receive(minimumIncompleteLength: 1, maximumLength: 16 * 1024) { [weak self] data, _, isComplete, error in
|
||||
guard let self else { return }
|
||||
do {
|
||||
if let data, let request = try parser.value.append(Array(data)) {
|
||||
// Signalling may block briefly (session setup); keep the HTTP queue free.
|
||||
DispatchQueue.global(qos: .userInitiated).async {
|
||||
self.respond(self.handler(request), on: connection)
|
||||
}
|
||||
return
|
||||
}
|
||||
} catch HTTPRequestParser.ParseError.tooLarge {
|
||||
respond(.text(413, "Zu groß."), on: connection)
|
||||
return
|
||||
} catch {
|
||||
respond(.text(400, "Ungültige Anfrage."), on: connection)
|
||||
return
|
||||
}
|
||||
if isComplete || error != nil { connection.cancel() } else { read(connection, parser: parser) }
|
||||
}
|
||||
}
|
||||
|
||||
private func respond(_ response: HTTPResponse, on connection: NWConnection) {
|
||||
connection.send(content: Data(response.serialized), completion: .contentProcessed { _ in connection.cancel() })
|
||||
}
|
||||
}
|
||||
|
||||
/// Per-connection parser state; only touched on the HTTP queue.
|
||||
private final class ParserBox: @unchecked Sendable {
|
||||
var value = HTTPRequestParser()
|
||||
}
|
||||
@ -0,0 +1,218 @@
|
||||
import Foundation
|
||||
import Network
|
||||
import OSLog
|
||||
|
||||
/// Media transport for all phones: one UDP port on the clinic interface, demultiplexed into sessions.
|
||||
/// Everything runs on one serial queue; public methods may be called from any thread.
|
||||
final class IntercomEngine: @unchecked Sendable {
|
||||
struct SessionInfo: Sendable, Equatable {
|
||||
let id: UUID
|
||||
let name: String
|
||||
let role: String
|
||||
let connected: Bool
|
||||
}
|
||||
|
||||
var onSessionsChanged: (@Sendable ([SessionInfo]) -> Void)?
|
||||
var onKeyFrameNeeded: (@Sendable () -> Void)?
|
||||
|
||||
let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.media", qos: .userInteractive)
|
||||
private let certificate: DTLSCertificate
|
||||
private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")
|
||||
|
||||
// queue only
|
||||
private var listener: NWListener?
|
||||
private var sessions: [UUID: PeerSession] = [:]
|
||||
private var connections: [UUID: NWConnection] = [:]
|
||||
private var timer: DispatchSourceTimer?
|
||||
|
||||
init(certificate: DTLSCertificate) {
|
||||
self.certificate = certificate
|
||||
}
|
||||
|
||||
/// Binds the media port; returns once listening (or throws).
|
||||
func start(interface: NWInterface, port: UInt16) async throws {
|
||||
let parameters = NWParameters.udp
|
||||
parameters.requiredInterface = interface
|
||||
parameters.allowLocalEndpointReuse = true
|
||||
let listener = try NWListener(using: parameters, on: NWEndpoint.Port(rawValue: port) ?? .any)
|
||||
let once = OnceFlag()
|
||||
try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
|
||||
listener.stateUpdateHandler = { [log] state in
|
||||
switch state {
|
||||
case .ready:
|
||||
if once.claim() { continuation.resume() }
|
||||
case .failed(let error):
|
||||
log.error("Media port failed: \(error.localizedDescription, privacy: .public)")
|
||||
if once.claim() { continuation.resume(throwing: error) }
|
||||
default: break
|
||||
}
|
||||
}
|
||||
listener.newConnectionHandler = { [weak self] connection in self?.accept(connection) }
|
||||
listener.start(queue: queue)
|
||||
queue.async { self.listener = listener }
|
||||
}
|
||||
queue.async { self.startTimer() }
|
||||
}
|
||||
|
||||
func stop() {
|
||||
queue.sync {
|
||||
timer?.cancel()
|
||||
timer = nil
|
||||
listener?.cancel()
|
||||
listener = nil
|
||||
connections.values.forEach { $0.cancel() }
|
||||
connections = [:]
|
||||
sessions = [:]
|
||||
}
|
||||
onSessionsChanged?([])
|
||||
}
|
||||
|
||||
/// Creates a session for a browser offer and returns its id and the SDP answer.
|
||||
func createSession(offer: String, name: String, role: String, address: String, port: UInt16) throws -> (UUID, String)? {
|
||||
let accepted = try PeerSession.accept(offerSDP: offer, name: name, role: role, certificate: certificate,
|
||||
address: address, port: port)
|
||||
guard let accepted else { return nil }
|
||||
queue.sync {
|
||||
sessions[accepted.session.id] = accepted.session
|
||||
publish()
|
||||
}
|
||||
log.info("Session for \(name, privacy: .public) (\(role, privacy: .public)) created")
|
||||
return (accepted.session.id, accepted.answer)
|
||||
}
|
||||
|
||||
func removeSession(_ id: UUID) {
|
||||
queue.async {
|
||||
self.sessions[id] = nil
|
||||
self.connections.removeValue(forKey: id)?.cancel()
|
||||
self.publish()
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends one Opus packet (20 ms) to every connected phone.
|
||||
func sendAudio(opus: [UInt8], timestamp: UInt32) {
|
||||
queue.async {
|
||||
for (id, session) in self.sessions {
|
||||
guard let packet = session.audioPacket(opus: opus, timestamp: timestamp) else { continue }
|
||||
self.connections[id]?.send(content: Data(packet), completion: .idempotent)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends one H.264 access unit to every connected phone.
|
||||
func sendVideo(nalUnits: [[UInt8]], timestamp: UInt32) {
|
||||
let payloads = H264Packetizer.payloads(for: nalUnits)
|
||||
queue.async {
|
||||
for (id, session) in self.sessions {
|
||||
guard let connection = self.connections[id] else { continue }
|
||||
for packet in session.videoPackets(payloads: payloads, timestamp: timestamp) {
|
||||
connection.send(content: Data(packet), completion: .idempotent)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - queue
|
||||
|
||||
private func accept(_ connection: NWConnection) {
|
||||
connection.start(queue: queue)
|
||||
receive(on: connection)
|
||||
}
|
||||
|
||||
private func receive(on connection: NWConnection) {
|
||||
connection.receiveMessage { [weak self] data, _, _, error in
|
||||
guard let self else { return }
|
||||
if let data, !data.isEmpty { self.handle(Array(data), from: connection) }
|
||||
if error == nil { self.receive(on: connection) } else { connection.cancel() }
|
||||
}
|
||||
}
|
||||
|
||||
private func handle(_ packet: [UInt8], from connection: NWConnection) {
|
||||
switch PeerSession.classify(packet) {
|
||||
case .stun:
|
||||
guard let message = STUN.parse(packet),
|
||||
case .hostPort(let host, let port) = connection.endpoint else { return }
|
||||
let address: [UInt8]
|
||||
switch host {
|
||||
case .ipv4(let v4): address = Array(v4.rawValue)
|
||||
case .ipv6(let v6):
|
||||
let raw = Array(v6.rawValue)
|
||||
// IPv4-mapped (::ffff:a.b.c.d) on a dual-stack socket.
|
||||
address = raw.prefix(12).elementsEqual([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF]) ? Array(raw.suffix(4)) : raw
|
||||
default: return
|
||||
}
|
||||
let localUfrag = message.username?.split(separator: ":").first.map(String.init)
|
||||
guard let session = sessions.values.first(where: { $0.localUfrag == localUfrag }),
|
||||
let reply = session.answerSTUN(packet, message: message, address: address, port: port.rawValue) else { return }
|
||||
if connections[session.id] !== connection {
|
||||
connections[session.id] = connection
|
||||
log.info("\(session.name, privacy: .public) reachable via \(String(describing: connection.endpoint), privacy: .public)")
|
||||
}
|
||||
connection.send(content: Data(reply), completion: .idempotent)
|
||||
|
||||
case .dtls:
|
||||
guard let (id, session) = session(for: connection) else { return }
|
||||
let result = session.receiveDTLS(packet)
|
||||
for reply in result.replies { connection.send(content: Data(reply), completion: .idempotent) }
|
||||
if result.becameReady {
|
||||
log.info("\(session.name, privacy: .public) connected (DTLS-SRTP)")
|
||||
publish()
|
||||
onKeyFrameNeeded?()
|
||||
} else if case .failed(let reason) = session.dtlsState {
|
||||
log.error("DTLS with \(session.name, privacy: .public) failed: \(reason, privacy: .public)")
|
||||
sessions[id] = nil
|
||||
connections[id] = nil
|
||||
publish()
|
||||
}
|
||||
|
||||
case .rtcp:
|
||||
if let (_, session) = session(for: connection), session.receiveRTCP(packet) { onKeyFrameNeeded?() }
|
||||
|
||||
case .other:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
private func session(for connection: NWConnection) -> (UUID, PeerSession)? {
|
||||
guard let id = connections.first(where: { $0.value === connection })?.key, let session = sessions[id] else { return nil }
|
||||
return (id, session)
|
||||
}
|
||||
|
||||
private func startTimer() {
|
||||
let timer = DispatchSource.makeTimerSource(queue: queue)
|
||||
timer.schedule(deadline: .now() + 1, repeating: 1)
|
||||
timer.setEventHandler { [weak self] in self?.tick() }
|
||||
timer.resume()
|
||||
self.timer = timer
|
||||
}
|
||||
|
||||
private func tick() {
|
||||
let now = Date()
|
||||
var changed = false
|
||||
for (id, session) in sessions {
|
||||
if session.isExpired(now: now) {
|
||||
log.info("\(session.name, privacy: .public) gone (no consent)")
|
||||
sessions[id] = nil
|
||||
connections.removeValue(forKey: id)?.cancel()
|
||||
changed = true
|
||||
continue
|
||||
}
|
||||
if let connection = connections[id] {
|
||||
for report in session.senderReports(now: now) { connection.send(content: Data(report), completion: .idempotent) }
|
||||
}
|
||||
}
|
||||
if changed { publish() }
|
||||
}
|
||||
|
||||
private func publish() {
|
||||
let infos = sessions.values
|
||||
.sorted { $0.created < $1.created }
|
||||
.map { SessionInfo(id: $0.id, name: $0.name, role: $0.role, connected: $0.isMediaReady) }
|
||||
onSessionsChanged?(infos)
|
||||
}
|
||||
}
|
||||
|
||||
final class OnceFlag: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var done = false
|
||||
func claim() -> Bool { lock.withLock { defer { done = true }; return !done } }
|
||||
}
|
||||
@ -0,0 +1,238 @@
|
||||
import CoreVideo
|
||||
import Darwin
|
||||
import Foundation
|
||||
import Network
|
||||
import Observation
|
||||
import OSLog
|
||||
|
||||
/// Intercom stage 1 (M4): phones on the clinic interface open a page, hear talkback and see PGM.
|
||||
/// Web server and media port bind only to the chosen interface.
|
||||
@MainActor @Observable
|
||||
public final class IntercomServer {
|
||||
public struct Participant: Identifiable, Equatable, Sendable {
|
||||
public let id: UUID
|
||||
public let name: String
|
||||
public let role: String
|
||||
public let connected: Bool
|
||||
}
|
||||
|
||||
public static let httpPort: UInt16 = 8080
|
||||
public static let mediaPort: UInt16 = 50_000
|
||||
|
||||
public private(set) var participants: [Participant] = []
|
||||
/// Address phones open (also encoded in the QR code).
|
||||
public private(set) var url: URL?
|
||||
public private(set) var error: String?
|
||||
public private(set) var isRunning = false
|
||||
|
||||
/// Thread-safe media input (PGM frames, talkback audio).
|
||||
public nonisolated let input: IntercomInput
|
||||
|
||||
private var engine: IntercomEngine?
|
||||
private var http: HTTPServer?
|
||||
private var retryTask: Task<Void, Never>?
|
||||
private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")
|
||||
|
||||
public init() {
|
||||
input = IntercomInput()
|
||||
}
|
||||
|
||||
public func start(interfaceName: String) async {
|
||||
stop()
|
||||
guard let interface = await Self.interface(named: interfaceName), let address = Self.ipv4Address(of: interfaceName) else {
|
||||
error = "Klinik-Schnittstelle \(interfaceName) hat keine Adresse. Neuer Versuch in 5 s."
|
||||
log.error("Intercom: interface \(interfaceName, privacy: .public) unavailable")
|
||||
scheduleRetry(interfaceName)
|
||||
return
|
||||
}
|
||||
do {
|
||||
let engine = IntercomEngine(certificate: try DTLSCertificate())
|
||||
engine.onSessionsChanged = { [weak self, input] infos in
|
||||
input.hasListeners = infos.contains { $0.connected }
|
||||
Task { @MainActor in
|
||||
self?.participants = infos.map { Participant(id: $0.id, name: $0.name, role: $0.role, connected: $0.connected) }
|
||||
}
|
||||
}
|
||||
engine.onKeyFrameNeeded = { [input] in input.requestKeyFrame() }
|
||||
try await engine.start(interface: interface, port: Self.mediaPort)
|
||||
|
||||
let mediaPort = Self.mediaPort
|
||||
let http = HTTPServer { [weak engine] request in
|
||||
guard let engine else { return .text(503, "Musubi startet neu.") }
|
||||
return Self.route(request, engine: engine, address: address, mediaPort: mediaPort)
|
||||
}
|
||||
try await http.start(interface: interface, port: Self.httpPort)
|
||||
|
||||
input.attach(engine)
|
||||
self.engine = engine
|
||||
self.http = http
|
||||
url = URL(string: "http://\(address):\(Self.httpPort)/")
|
||||
isRunning = true
|
||||
error = nil
|
||||
log.info("Intercom on \(interfaceName, privacy: .public) at \(address, privacy: .public)")
|
||||
} catch {
|
||||
stop()
|
||||
self.error = "Intercom lässt sich auf \(interfaceName) nicht starten. Neuer Versuch in 5 s."
|
||||
log.error("Intercom start failed: \(error.localizedDescription, privacy: .public)")
|
||||
scheduleRetry(interfaceName)
|
||||
}
|
||||
}
|
||||
|
||||
public func stop() {
|
||||
retryTask?.cancel()
|
||||
retryTask = nil
|
||||
input.attach(nil)
|
||||
http?.stop()
|
||||
http = nil
|
||||
engine?.stop()
|
||||
engine = nil
|
||||
participants = []
|
||||
url = nil
|
||||
isRunning = false
|
||||
}
|
||||
|
||||
private func scheduleRetry(_ interfaceName: String) {
|
||||
retryTask = Task { [weak self] in
|
||||
try? await Task.sleep(for: .seconds(5))
|
||||
guard !Task.isCancelled else { return }
|
||||
await self?.start(interfaceName: interfaceName)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Signalling
|
||||
|
||||
private struct OfferRequest: Decodable { let name: String; let role: String; let sdp: String }
|
||||
private struct OfferResponse: Encodable { let id: UUID; let sdp: String }
|
||||
private struct LeaveRequest: Decodable { let id: UUID }
|
||||
|
||||
nonisolated private static func route(_ request: HTTPRequest, engine: IntercomEngine, address: String, mediaPort: UInt16) -> HTTPResponse {
|
||||
switch (request.method, request.path) {
|
||||
case ("GET", "/"), ("GET", "/index.html"):
|
||||
return HTTPResponse(status: 200, contentType: "text/html; charset=utf-8", body: Array(PhonePage.html.utf8))
|
||||
case ("POST", "/api/offer"):
|
||||
guard let offer = try? JSONDecoder().decode(OfferRequest.self, from: Data(request.body)) else {
|
||||
return .text(400, "Angebot unlesbar.")
|
||||
}
|
||||
let name = String(offer.name.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40))
|
||||
let role = String(offer.role.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40))
|
||||
guard !name.isEmpty, !role.isEmpty else { return .text(400, "Name und Rolle fehlen.") }
|
||||
do {
|
||||
guard let (id, sdp) = try engine.createSession(offer: offer.sdp, name: name, role: role,
|
||||
address: address, port: mediaPort) else {
|
||||
return .text(400, "Dein Browser kann weder Opus noch H.264 empfangen.")
|
||||
}
|
||||
return .json(OfferResponse(id: id, sdp: sdp))
|
||||
} catch {
|
||||
return .text(400, "Angebot unvollständig.")
|
||||
}
|
||||
case ("POST", "/api/leave"):
|
||||
if let leave = try? JSONDecoder().decode(LeaveRequest.self, from: Data(request.body)) { engine.removeSession(leave.id) }
|
||||
return HTTPResponse(status: 204, contentType: "text/plain", body: [])
|
||||
case (_, "/"), (_, "/api/offer"), (_, "/api/leave"):
|
||||
return .text(405, "Methode nicht erlaubt.")
|
||||
default:
|
||||
return .text(404, "Nicht gefunden.")
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - Interfaces
|
||||
|
||||
private static func interface(named name: String) async -> NWInterface? {
|
||||
await withCheckedContinuation { continuation in
|
||||
let monitor = NWPathMonitor()
|
||||
let once = OnceFlag()
|
||||
monitor.pathUpdateHandler = { path in
|
||||
guard once.claim() else { return }
|
||||
monitor.cancel()
|
||||
continuation.resume(returning: path.availableInterfaces.first { $0.name == name })
|
||||
}
|
||||
monitor.start(queue: .global(qos: .utility))
|
||||
}
|
||||
}
|
||||
|
||||
/// First IPv4 address of a BSD interface (e.g. "en0").
|
||||
nonisolated static func ipv4Address(of interfaceName: String) -> String? {
|
||||
var list: UnsafeMutablePointer<ifaddrs>?
|
||||
guard getifaddrs(&list) == 0, let first = list else { return nil }
|
||||
defer { freeifaddrs(list) }
|
||||
for entry in sequence(first: first, next: { $0.pointee.ifa_next }) {
|
||||
let ifa = entry.pointee
|
||||
guard String(cString: ifa.ifa_name) == interfaceName, let addr = ifa.ifa_addr,
|
||||
addr.pointee.sa_family == UInt8(AF_INET) else { continue }
|
||||
var host = [CChar](repeating: 0, count: Int(NI_MAXHOST))
|
||||
guard getnameinfo(addr, socklen_t(addr.pointee.sa_len), &host, socklen_t(host.count), nil, 0, NI_NUMERICHOST) == 0 else { continue }
|
||||
return String(decoding: host.prefix { $0 != 0 }.map { UInt8(bitPattern: $0) }, as: UTF8.self)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
/// Media input for the intercom; safe to call from any thread.
|
||||
public final class IntercomInput: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var engine: IntercomEngine?
|
||||
private var listeners = false
|
||||
private var talkback = TalkbackEncoder()
|
||||
private let preview = PreviewEncoder()
|
||||
private var resamplePosition: Double = 0
|
||||
private var resampleLast: Float = 0
|
||||
private var scratch: [Float] = []
|
||||
|
||||
init() {
|
||||
preview.onAccessUnit = { [weak self] nalUnits, timestamp in
|
||||
self?.currentEngine?.sendVideo(nalUnits: nalUnits, timestamp: timestamp)
|
||||
}
|
||||
}
|
||||
|
||||
var hasListeners: Bool {
|
||||
get { lock.withLock { listeners } }
|
||||
set { lock.withLock { listeners = newValue } }
|
||||
}
|
||||
|
||||
private var currentEngine: IntercomEngine? { lock.withLock { engine } }
|
||||
|
||||
func attach(_ engine: IntercomEngine?) {
|
||||
lock.withLock {
|
||||
self.engine = engine
|
||||
listeners = false
|
||||
}
|
||||
}
|
||||
|
||||
func requestKeyFrame() { preview.requestKeyFrame() }
|
||||
|
||||
/// One PGM frame (any size, BGRA or UYVY). Dropped while no phone is connected.
|
||||
public func pushVideo(_ pixelBuffer: CVPixelBuffer) {
|
||||
guard hasListeners else { return }
|
||||
preview.encode(pixelBuffer)
|
||||
}
|
||||
|
||||
/// Talkback audio, mono, any sample rate. Pass silence while the talk key is up so the stream keeps its timing.
|
||||
/// Call from one thread only (not the audio render thread).
|
||||
public func pushTalkback(_ samples: UnsafeBufferPointer<Float>, sampleRate: Double) {
|
||||
guard let engine = currentEngine, hasListeners, let talkback else { return }
|
||||
let packets: [([UInt8], UInt32)]
|
||||
if sampleRate == TalkbackEncoder.sampleRate {
|
||||
packets = talkback.append(samples)
|
||||
} else {
|
||||
resample(samples, from: sampleRate)
|
||||
packets = scratch.withUnsafeBufferPointer { talkback.append($0) }
|
||||
}
|
||||
for (packet, timestamp) in packets { engine.sendAudio(opus: packet, timestamp: timestamp) }
|
||||
}
|
||||
|
||||
/// Linear interpolation to 48 kHz; good enough for speech.
|
||||
private func resample(_ input: UnsafeBufferPointer<Float>, from rate: Double) {
|
||||
scratch.removeAll(keepingCapacity: true)
|
||||
let step = rate / TalkbackEncoder.sampleRate
|
||||
var position = resamplePosition
|
||||
while position < Double(input.count) {
|
||||
let index = Int(position)
|
||||
let fraction = Float(position - Double(index))
|
||||
let previous = index == 0 ? resampleLast : input[index - 1]
|
||||
scratch.append(previous + (input[index] - previous) * fraction)
|
||||
position += step
|
||||
}
|
||||
resamplePosition = position - Double(input.count)
|
||||
resampleLast = input.last ?? resampleLast
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,160 @@
|
||||
import CoreMedia
|
||||
import QuartzCore
|
||||
import CoreVideo
|
||||
import Foundation
|
||||
import OSLog
|
||||
import VideoToolbox
|
||||
|
||||
/// Encodes the PGM preview for phones: scaled down, 25 fps, H.264 constrained baseline, no B-frames.
|
||||
/// `encode` may be called from the NDI receive thread; work happens on an own queue.
|
||||
final class PreviewEncoder: @unchecked Sendable {
|
||||
/// Called with the NAL units of one access unit and its 90 kHz timestamp.
|
||||
var onAccessUnit: (@Sendable ([[UInt8]], UInt32) -> Void)?
|
||||
|
||||
let width: Int32
|
||||
let height: Int32
|
||||
let frameRate: Double
|
||||
private let bitRate: Int
|
||||
private let queue = DispatchQueue(label: "studio.zielke.musubi.intercom.video", qos: .userInitiated)
|
||||
private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")
|
||||
|
||||
// queue only
|
||||
private var session: VTCompressionSession?
|
||||
private var transfer: VTPixelTransferSession?
|
||||
private var pool: CVPixelBufferPool?
|
||||
private var forceKeyFrame = true
|
||||
private var lastFrameTime: Double = -1
|
||||
private var busy = false
|
||||
private let start = CACurrentMediaTime()
|
||||
|
||||
init(width: Int32 = 960, height: Int32 = 540, frameRate: Double = 25, bitRate: Int = 1_200_000) {
|
||||
self.width = width
|
||||
self.height = height
|
||||
self.frameRate = frameRate
|
||||
self.bitRate = bitRate
|
||||
}
|
||||
|
||||
deinit {
|
||||
if let session { VTCompressionSessionInvalidate(session) }
|
||||
if let transfer { VTPixelTransferSessionInvalidate(transfer) }
|
||||
}
|
||||
|
||||
func requestKeyFrame() {
|
||||
queue.async { self.forceKeyFrame = true }
|
||||
}
|
||||
|
||||
/// Offers a frame; frames above the target rate, or while the encoder is still busy, are dropped.
|
||||
func encode(_ pixelBuffer: CVPixelBuffer) {
|
||||
let now = CACurrentMediaTime() - start
|
||||
// CVPixelBuffer is immutable once handed over by the receiver.
|
||||
nonisolated(unsafe) let pixelBuffer = pixelBuffer
|
||||
queue.async {
|
||||
guard !self.busy, now - self.lastFrameTime >= 1 / self.frameRate - 0.004 else { return }
|
||||
self.lastFrameTime = now
|
||||
self.encodeOnQueue(pixelBuffer, time: now)
|
||||
}
|
||||
}
|
||||
|
||||
private func encodeOnQueue(_ source: CVPixelBuffer, time: Double) {
|
||||
guard let session = session ?? makeSession(), let scaled = scale(source) else { return }
|
||||
var options: CFDictionary?
|
||||
if forceKeyFrame {
|
||||
options = [kVTEncodeFrameOptionKey_ForceKeyFrame: true] as CFDictionary
|
||||
forceKeyFrame = false
|
||||
}
|
||||
busy = true
|
||||
let pts = CMTime(seconds: time, preferredTimescale: 90_000)
|
||||
let status = VTCompressionSessionEncodeFrame(session, imageBuffer: scaled, presentationTimeStamp: pts, duration: .invalid,
|
||||
frameProperties: options, infoFlagsOut: nil) { [weak self] status, _, sample in
|
||||
guard let self else { return }
|
||||
self.queue.async { self.busy = false }
|
||||
guard status == noErr, let sample else { return }
|
||||
self.emit(sample, timestamp: UInt32(truncatingIfNeeded: Int64(time * 90_000)))
|
||||
}
|
||||
if status != noErr {
|
||||
busy = false
|
||||
log.error("Preview encode failed: \(status)")
|
||||
}
|
||||
}
|
||||
|
||||
private func makeSession() -> VTCompressionSession? {
|
||||
var session: VTCompressionSession?
|
||||
let status = VTCompressionSessionCreate(allocator: nil, width: width, height: height, codecType: kCMVideoCodecType_H264,
|
||||
encoderSpecification: nil, imageBufferAttributes: nil, compressedDataAllocator: nil,
|
||||
outputCallback: nil, refcon: nil, compressionSessionOut: &session)
|
||||
guard status == noErr, let session else {
|
||||
log.error("Preview encoder unavailable: \(status)")
|
||||
return nil
|
||||
}
|
||||
let properties: [CFString: Any] = [
|
||||
kVTCompressionPropertyKey_RealTime: true,
|
||||
kVTCompressionPropertyKey_ProfileLevel: kVTProfileLevel_H264_ConstrainedBaseline_AutoLevel,
|
||||
kVTCompressionPropertyKey_AllowFrameReordering: false,
|
||||
kVTCompressionPropertyKey_AverageBitRate: bitRate,
|
||||
kVTCompressionPropertyKey_ExpectedFrameRate: frameRate,
|
||||
kVTCompressionPropertyKey_MaxKeyFrameInterval: Int(frameRate * 2),
|
||||
kVTCompressionPropertyKey_MaxKeyFrameIntervalDuration: 2,
|
||||
]
|
||||
for (key, value) in properties { VTSessionSetProperty(session, key: key, value: value as CFTypeRef) }
|
||||
VTCompressionSessionPrepareToEncodeFrames(session)
|
||||
self.session = session
|
||||
return session
|
||||
}
|
||||
|
||||
/// Scales any NDI frame (BGRA or UYVY, any size) into a pooled NV12 buffer of the preview size.
|
||||
private func scale(_ source: CVPixelBuffer) -> CVPixelBuffer? {
|
||||
if transfer == nil {
|
||||
var newTransfer: VTPixelTransferSession?
|
||||
VTPixelTransferSessionCreate(allocator: nil, pixelTransferSessionOut: &newTransfer)
|
||||
if let newTransfer {
|
||||
VTSessionSetProperty(newTransfer, key: kVTPixelTransferPropertyKey_ScalingMode, value: kVTScalingMode_Letterbox)
|
||||
}
|
||||
transfer = newTransfer
|
||||
}
|
||||
if pool == nil {
|
||||
let attributes: [CFString: Any] = [
|
||||
kCVPixelBufferPixelFormatTypeKey: kCVPixelFormatType_420YpCbCr8BiPlanarVideoRange,
|
||||
kCVPixelBufferWidthKey: width,
|
||||
kCVPixelBufferHeightKey: height,
|
||||
kCVPixelBufferIOSurfacePropertiesKey: [:] as CFDictionary,
|
||||
]
|
||||
CVPixelBufferPoolCreate(nil, nil, attributes as CFDictionary, &pool)
|
||||
}
|
||||
guard let transfer, let pool else { return nil }
|
||||
var target: CVPixelBuffer?
|
||||
CVPixelBufferPoolCreatePixelBuffer(nil, pool, &target)
|
||||
guard let target, VTPixelTransferSessionTransferImage(transfer, from: source, to: target) == noErr else { return nil }
|
||||
return target
|
||||
}
|
||||
|
||||
private func emit(_ sample: CMSampleBuffer, timestamp: UInt32) {
|
||||
guard let block = CMSampleBufferGetDataBuffer(sample) else { return }
|
||||
var length = 0
|
||||
var pointer: UnsafeMutablePointer<CChar>?
|
||||
guard CMBlockBufferGetDataPointer(block, atOffset: 0, lengthAtOffsetOut: nil, totalLengthOut: &length,
|
||||
dataPointerOut: &pointer) == noErr, let pointer else { return }
|
||||
let avcc = [UInt8](UnsafeRawBufferPointer(start: pointer, count: length))
|
||||
var nalUnits = H264Packetizer.nalUnits(avcc: avcc)
|
||||
|
||||
let attachments = CMSampleBufferGetSampleAttachmentsArray(sample, createIfNecessary: false) as? [[CFString: Any]]
|
||||
let isKeyFrame = !(attachments?.first?[kCMSampleAttachmentKey_NotSync] as? Bool ?? false)
|
||||
if isKeyFrame, let format = CMSampleBufferGetFormatDescription(sample) {
|
||||
nalUnits = Self.parameterSets(format) + nalUnits
|
||||
}
|
||||
onAccessUnit?(nalUnits, timestamp)
|
||||
}
|
||||
|
||||
private static func parameterSets(_ format: CMFormatDescription) -> [[UInt8]] {
|
||||
var count = 0
|
||||
CMVideoFormatDescriptionGetH264ParameterSetAtIndex(format, parameterSetIndex: 0, parameterSetPointerOut: nil,
|
||||
parameterSetSizeOut: nil, parameterSetCountOut: &count, nalUnitHeaderLengthOut: nil)
|
||||
return (0..<count).compactMap { index in
|
||||
var pointer: UnsafePointer<UInt8>?
|
||||
var size = 0
|
||||
guard CMVideoFormatDescriptionGetH264ParameterSetAtIndex(format, parameterSetIndex: index, parameterSetPointerOut: &pointer,
|
||||
parameterSetSizeOut: &size, parameterSetCountOut: nil,
|
||||
nalUnitHeaderLengthOut: nil) == noErr, let pointer else { return nil }
|
||||
return Array(UnsafeBufferPointer(start: pointer, count: size))
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,90 @@
|
||||
import AudioToolbox
|
||||
import Foundation
|
||||
import OSLog
|
||||
|
||||
/// Encodes mono 48 kHz Float32 into 20 ms Opus packets with AudioToolbox.
|
||||
/// Not thread-safe; the caller feeds it from one thread.
|
||||
final class TalkbackEncoder {
|
||||
static let sampleRate = 48_000.0
|
||||
static let frameSamples = 960 // 20 ms
|
||||
|
||||
private var converter: AudioConverterRef?
|
||||
private var pending: [Float] = []
|
||||
private var timestamp: UInt32 = .random(in: 0...UInt32.max)
|
||||
private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")
|
||||
|
||||
init?(bitRate: UInt32 = 32_000) {
|
||||
var input = AudioStreamBasicDescription(
|
||||
mSampleRate: Self.sampleRate, mFormatID: kAudioFormatLinearPCM,
|
||||
mFormatFlags: kAudioFormatFlagIsFloat | kAudioFormatFlagIsPacked,
|
||||
mBytesPerPacket: 4, mFramesPerPacket: 1, mBytesPerFrame: 4, mChannelsPerFrame: 1, mBitsPerChannel: 32, mReserved: 0)
|
||||
var output = AudioStreamBasicDescription(
|
||||
mSampleRate: Self.sampleRate, mFormatID: kAudioFormatOpus, mFormatFlags: 0,
|
||||
mBytesPerPacket: 0, mFramesPerPacket: UInt32(Self.frameSamples), mBytesPerFrame: 0, mChannelsPerFrame: 1,
|
||||
mBitsPerChannel: 0, mReserved: 0)
|
||||
let status = AudioConverterNew(&input, &output, &converter)
|
||||
guard status == noErr, let converter else {
|
||||
log.error("Opus encoder unavailable: \(status)")
|
||||
return nil
|
||||
}
|
||||
var rate = bitRate
|
||||
AudioConverterSetProperty(converter, kAudioConverterEncodeBitRate, UInt32(MemoryLayout<UInt32>.size), &rate)
|
||||
}
|
||||
|
||||
deinit { if let converter { AudioConverterDispose(converter) } }
|
||||
|
||||
/// Appends mono samples at 48 kHz; returns (packet, RTP timestamp) for every full 20 ms.
|
||||
func append(_ samples: UnsafeBufferPointer<Float>) -> [([UInt8], UInt32)] {
|
||||
pending += samples
|
||||
var out: [([UInt8], UInt32)] = []
|
||||
while pending.count >= Self.frameSamples {
|
||||
let frame = Array(pending.prefix(Self.frameSamples))
|
||||
pending.removeFirst(Self.frameSamples)
|
||||
if let packet = encode(frame) { out.append((packet, timestamp)) }
|
||||
timestamp &+= UInt32(Self.frameSamples)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/// Input for one converter call; owns stable memory the converter may read after the callback returns.
|
||||
private final class Feed {
|
||||
let samples: UnsafeMutablePointer<Float>
|
||||
let count: Int
|
||||
var consumed = false
|
||||
init(_ frame: [Float]) {
|
||||
count = frame.count
|
||||
samples = .allocate(capacity: count)
|
||||
samples.initialize(from: frame, count: count)
|
||||
}
|
||||
deinit { samples.deallocate() }
|
||||
}
|
||||
|
||||
private func encode(_ frame: [Float]) -> [UInt8]? {
|
||||
guard let converter else { return nil }
|
||||
let feed = Feed(frame)
|
||||
var packetSize: UInt32 = 1500
|
||||
var outBytes = [UInt8](repeating: 0, count: Int(packetSize))
|
||||
var packets: UInt32 = 1
|
||||
var description = AudioStreamPacketDescription()
|
||||
let status = withExtendedLifetime(feed) { outBytes.withUnsafeMutableBytes { raw -> OSStatus in
|
||||
var list = AudioBufferList(mNumberBuffers: 1, mBuffers: AudioBuffer(mNumberChannels: 1, mDataByteSize: packetSize, mData: raw.baseAddress))
|
||||
let status = AudioConverterFillComplexBuffer(converter, { _, ioPackets, ioData, _, userData in
|
||||
let feed = Unmanaged<Feed>.fromOpaque(userData!).takeUnretainedValue()
|
||||
guard !feed.consumed else {
|
||||
ioPackets.pointee = 0
|
||||
return 1 // no more data for this call
|
||||
}
|
||||
feed.consumed = true
|
||||
ioPackets.pointee = UInt32(feed.count)
|
||||
ioData.pointee.mBuffers.mData = UnsafeMutableRawPointer(feed.samples)
|
||||
ioData.pointee.mBuffers.mDataByteSize = UInt32(feed.count * 4)
|
||||
ioData.pointee.mBuffers.mNumberChannels = 1
|
||||
return noErr
|
||||
}, Unmanaged.passUnretained(feed).toOpaque(), &packets, &list, &description)
|
||||
packetSize = list.mBuffers.mDataByteSize
|
||||
return status
|
||||
} }
|
||||
guard (status == noErr || status == 1), packets == 1, packetSize > 0 else { return nil }
|
||||
return Array(outBytes.prefix(Int(packetSize)))
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,151 @@
|
||||
/// The page phones open from the QR code. Colors follow design/tokens.json (dark).
|
||||
enum PhonePage {
|
||||
static let html = #"""
|
||||
<!doctype html>
|
||||
<html lang="de">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover">
|
||||
<meta name="theme-color" content="#14151B">
|
||||
<title>Musubi</title>
|
||||
<style>
|
||||
:root { --sumi:#14151B; --kasumi:#1E2029; --usuzumi:#2B2E3A; --washi:#ECE9E2; --nezumi:#8E8C96; --ai:#8A99F2; --moegi:#3DBE7A; --kohaku:#E8A23A; }
|
||||
* { box-sizing: border-box; }
|
||||
html, body { margin: 0; background: var(--sumi); color: var(--washi);
|
||||
font: 16px/1.4 -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; }
|
||||
main { max-width: 720px; margin: 0 auto; padding: max(16px, env(safe-area-inset-top)) 16px 24px; }
|
||||
h1 { font-size: 20px; font-weight: 600; margin: 8px 0 2px; }
|
||||
.sub { color: var(--nezumi); margin: 0 0 16px; }
|
||||
.card { background: var(--kasumi); border: 1px solid var(--usuzumi); border-radius: 12px; padding: 16px; }
|
||||
label { display: block; color: var(--nezumi); font-size: 14px; margin: 0 0 6px; }
|
||||
input { width: 100%; font: inherit; color: var(--washi); background: var(--sumi); border: 1px solid var(--usuzumi);
|
||||
border-radius: 8px; padding: 12px; margin: 0 0 16px; }
|
||||
input:focus { outline: 2px solid var(--ai); outline-offset: -1px; }
|
||||
button { width: 100%; font: inherit; font-weight: 600; border: 0; border-radius: 8px; padding: 14px; cursor: pointer; }
|
||||
.primary { background: var(--ai); color: var(--sumi); }
|
||||
.secondary { background: var(--usuzumi); color: var(--washi); margin-top: 8px; }
|
||||
video { width: 100%; aspect-ratio: 16 / 9; background: #000; border-radius: 12px; display: block; }
|
||||
.status { display: flex; align-items: center; gap: 8px; color: var(--nezumi); margin: 12px 0; font-size: 14px; }
|
||||
.dot { width: 8px; height: 8px; border-radius: 4px; background: var(--usuzumi); flex: none; }
|
||||
.dot.ok { background: var(--moegi); }
|
||||
.dot.warn { background: var(--kohaku); }
|
||||
.who { color: var(--washi); }
|
||||
[hidden] { display: none !important; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<main>
|
||||
<h1>Musubi</h1>
|
||||
<p class="sub">Talkback hören, Sendung sehen.</p>
|
||||
|
||||
<form id="join" class="card">
|
||||
<label for="name">Dein Name</label>
|
||||
<input id="name" autocomplete="name" required placeholder="z. B. Anna">
|
||||
<label for="role">Deine Rolle</label>
|
||||
<input id="role" list="roles" required placeholder="z. B. Kamera 2">
|
||||
<datalist id="roles"><option>Kamera 1</option><option>Kamera 2</option><option>Kamera 3</option><option>Technik</option></datalist>
|
||||
<button class="primary" type="submit">Verbinden</button>
|
||||
</form>
|
||||
|
||||
<section id="live" hidden>
|
||||
<video id="video" playsinline autoplay></video>
|
||||
<div class="status"><span id="dot" class="dot"></span><span id="status">Verbinde …</span></div>
|
||||
<button id="unmute" class="primary" hidden>Ton einschalten</button>
|
||||
<button id="leave" class="secondary">Trennen</button>
|
||||
</section>
|
||||
</main>
|
||||
<script>
|
||||
const $ = (s) => document.querySelector(s);
|
||||
let pc = null, sessionId = null, wantConnected = false, retryTimer = null;
|
||||
|
||||
try {
|
||||
$("#name").value = localStorage.getItem("musubi.name") || "";
|
||||
$("#role").value = localStorage.getItem("musubi.role") || "";
|
||||
} catch {}
|
||||
|
||||
function setStatus(text, kind) {
|
||||
$("#status").textContent = text;
|
||||
$("#dot").className = "dot" + (kind ? " " + kind : "");
|
||||
}
|
||||
|
||||
async function connect() {
|
||||
clearTimeout(retryTimer);
|
||||
const name = $("#name").value.trim(), role = $("#role").value.trim();
|
||||
setStatus("Verbinde …", "warn");
|
||||
pc = new RTCPeerConnection({ iceServers: [] });
|
||||
const current = pc;
|
||||
pc.addTransceiver("audio", { direction: "recvonly" });
|
||||
pc.addTransceiver("video", { direction: "recvonly" });
|
||||
const stream = new MediaStream();
|
||||
pc.ontrack = (e) => {
|
||||
stream.addTrack(e.track);
|
||||
const video = $("#video");
|
||||
if (video.srcObject !== stream) video.srcObject = stream;
|
||||
video.play().then(() => { $("#unmute").hidden = true; }).catch(() => { $("#unmute").hidden = false; });
|
||||
};
|
||||
pc.onconnectionstatechange = () => {
|
||||
if (current !== pc) return;
|
||||
const state = pc.connectionState;
|
||||
if (state === "connected") setStatus("Verbunden als " + name + " · " + role, "ok");
|
||||
if (state === "failed" || state === "disconnected" || state === "closed") reconnectSoon();
|
||||
};
|
||||
try {
|
||||
await pc.setLocalDescription(await pc.createOffer());
|
||||
const response = await fetch("/api/offer", {
|
||||
method: "POST", headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ name, role, sdp: pc.localDescription.sdp })
|
||||
});
|
||||
if (!response.ok) throw new Error(await response.text());
|
||||
const answer = await response.json();
|
||||
sessionId = answer.id;
|
||||
await pc.setRemoteDescription({ type: "answer", sdp: answer.sdp });
|
||||
} catch (error) {
|
||||
reconnectSoon(error && error.message);
|
||||
}
|
||||
}
|
||||
|
||||
function reconnectSoon(reason) {
|
||||
if (!wantConnected) return;
|
||||
setStatus((reason || "Verbindung weg.") + " Neuer Versuch in 3 s.", "warn");
|
||||
teardown();
|
||||
clearTimeout(retryTimer);
|
||||
retryTimer = setTimeout(() => { if (wantConnected) connect(); }, 3000);
|
||||
}
|
||||
|
||||
function teardown() {
|
||||
if (sessionId) navigator.sendBeacon("/api/leave", JSON.stringify({ id: sessionId }));
|
||||
sessionId = null;
|
||||
if (pc) { pc.onconnectionstatechange = null; pc.close(); pc = null; }
|
||||
}
|
||||
|
||||
$("#join").addEventListener("submit", (e) => {
|
||||
e.preventDefault();
|
||||
try {
|
||||
localStorage.setItem("musubi.name", $("#name").value.trim());
|
||||
localStorage.setItem("musubi.role", $("#role").value.trim());
|
||||
} catch {}
|
||||
wantConnected = true;
|
||||
$("#join").hidden = true;
|
||||
$("#live").hidden = false;
|
||||
$("#video").play().catch(() => {}); // unlock playback inside the tap
|
||||
connect();
|
||||
});
|
||||
|
||||
$("#unmute").addEventListener("click", () => {
|
||||
$("#video").play().then(() => { $("#unmute").hidden = true; }).catch(() => {});
|
||||
});
|
||||
|
||||
$("#leave").addEventListener("click", () => {
|
||||
wantConnected = false;
|
||||
clearTimeout(retryTimer);
|
||||
teardown();
|
||||
$("#live").hidden = true;
|
||||
$("#join").hidden = false;
|
||||
});
|
||||
|
||||
addEventListener("pagehide", teardown);
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
"""#
|
||||
}
|
||||
@ -0,0 +1,150 @@
|
||||
import Foundation
|
||||
|
||||
/// One phone: ICE-lite credentials, DTLS, SRTP and the two outgoing RTP streams.
|
||||
/// Owned and called by `IntercomEngine` on its queue only.
|
||||
final class PeerSession {
|
||||
struct Offer {
|
||||
let answer: String
|
||||
let session: PeerSession
|
||||
}
|
||||
|
||||
enum Kind { case stun, dtls, rtcp, other }
|
||||
|
||||
let id = UUID()
|
||||
let name: String
|
||||
let role: String
|
||||
let localUfrag: String
|
||||
let localPassword: String
|
||||
let remoteUfrag: String
|
||||
let created = Date()
|
||||
|
||||
private(set) var lastConsent: Date?
|
||||
private(set) var connectedSince: Date?
|
||||
private var dtls: DTLSServer
|
||||
private var srtp: SRTPContext?
|
||||
private var audio: RTPStream?
|
||||
private var video: RTPStream?
|
||||
private var lastAudioTimestamp: UInt32 = 0
|
||||
private var lastVideoTimestamp: UInt32 = 0
|
||||
private var reportsDue = Date()
|
||||
|
||||
var isMediaReady: Bool { srtp != nil }
|
||||
var dtlsState: DTLSServer.State { dtls.state }
|
||||
|
||||
private init(name: String, role: String, remoteUfrag: String, certificate: DTLSCertificate, remoteFingerprint: String) {
|
||||
self.name = name
|
||||
self.role = role
|
||||
self.remoteUfrag = remoteUfrag
|
||||
localUfrag = Self.token(8)
|
||||
localPassword = Self.token(24)
|
||||
dtls = DTLSServer(certificate: certificate, expectedFingerprint: remoteFingerprint)
|
||||
}
|
||||
|
||||
/// Accepts a browser offer and returns the answer; nil if the offer has nothing Musubi can send.
|
||||
static func accept(offerSDP: String, name: String, role: String, certificate: DTLSCertificate,
|
||||
address: String, port: UInt16) throws -> Offer? {
|
||||
let offer = try SDPOffer(offerSDP)
|
||||
let session = PeerSession(name: name, role: role, remoteUfrag: offer.iceUfrag, certificate: certificate,
|
||||
remoteFingerprint: offer.fingerprint)
|
||||
var tracks: [SDPAnswer.Track] = []
|
||||
for media in offer.media {
|
||||
let ssrc = UInt32.random(in: 1...UInt32.max)
|
||||
var payloadType: Int?
|
||||
if media.port != 0 {
|
||||
if media.kind == "audio", session.audio == nil, let pt = offer.opusPayloadType(in: media) {
|
||||
payloadType = pt
|
||||
session.audio = RTPStream(ssrc: ssrc, payloadType: UInt8(pt))
|
||||
} else if media.kind == "video", session.video == nil, let pt = offer.h264PayloadType(in: media) {
|
||||
payloadType = pt
|
||||
session.video = RTPStream(ssrc: ssrc, payloadType: UInt8(pt))
|
||||
}
|
||||
}
|
||||
tracks.append(.init(mid: media.mid, kind: media.kind, payloadType: payloadType, ssrc: ssrc))
|
||||
}
|
||||
guard session.audio != nil || session.video != nil else { return nil }
|
||||
let answer = SDPAnswer(iceUfrag: session.localUfrag, icePwd: session.localPassword,
|
||||
fingerprint: certificate.fingerprint, address: address, port: port, tracks: tracks)
|
||||
return Offer(answer: answer.text, session: session)
|
||||
}
|
||||
|
||||
static func classify(_ packet: [UInt8]) -> Kind {
|
||||
if STUN.looksLikeSTUN(packet) { return .stun }
|
||||
if DTLSServer.looksLikeDTLS(packet) { return .dtls }
|
||||
if RTCPFeedback.isRTCP(packet) { return .rtcp }
|
||||
return .other
|
||||
}
|
||||
|
||||
/// Answers a connectivity check if it carries this session's credentials.
|
||||
func answerSTUN(_ packet: [UInt8], message: STUN.Message, address: [UInt8], port: UInt16) -> [UInt8]? {
|
||||
guard message.type == STUN.bindingRequest, message.username == "\(localUfrag):\(remoteUfrag)",
|
||||
STUN.verify(packet, message: message, key: Array(localPassword.utf8)) else { return nil }
|
||||
lastConsent = Date()
|
||||
return STUN.bindingResponse(transactionID: message.transactionID, address: address, port: port,
|
||||
key: Array(localPassword.utf8))
|
||||
}
|
||||
|
||||
/// Feeds DTLS; returns datagrams to send and whether media just became ready.
|
||||
func receiveDTLS(_ packet: [UInt8]) -> (replies: [[UInt8]], becameReady: Bool) {
|
||||
let replies = dtls.handle(packet)
|
||||
guard srtp == nil, let keys = dtls.keyMaterial else { return (replies, false) }
|
||||
// DTLS server = answerer: we send with the server keys.
|
||||
srtp = SRTPContext(masterKey: keys.serverKey, masterSalt: keys.serverSalt)
|
||||
connectedSince = Date()
|
||||
return (replies, true)
|
||||
}
|
||||
|
||||
/// True if the browser asks for a key frame.
|
||||
func receiveRTCP(_ packet: [UInt8]) -> Bool {
|
||||
guard let srtpIn = inboundRTCP, let plain = srtpIn.unprotectRTCP(packet) else { return false }
|
||||
return RTCPFeedback.requestsKeyFrame(plain)
|
||||
}
|
||||
|
||||
private lazy var inboundRTCP: SRTPContext? = {
|
||||
guard let keys = dtls.keyMaterial else { return nil }
|
||||
return SRTPContext(masterKey: keys.clientKey, masterSalt: keys.clientSalt)
|
||||
}()
|
||||
|
||||
func audioPacket(opus: [UInt8], timestamp: UInt32) -> [UInt8]? {
|
||||
guard let srtp, audio != nil else { return nil }
|
||||
lastAudioTimestamp = timestamp
|
||||
return srtp.protectRTP(audio!.packet(timestamp: timestamp, marker: false, payload: opus))
|
||||
}
|
||||
|
||||
func videoPackets(payloads: [[UInt8]], timestamp: UInt32) -> [[UInt8]] {
|
||||
guard let srtp, video != nil else { return [] }
|
||||
lastVideoTimestamp = timestamp
|
||||
return payloads.enumerated().map { index, payload in
|
||||
srtp.protectRTP(video!.packet(timestamp: timestamp, marker: index == payloads.count - 1, payload: payload))
|
||||
}
|
||||
}
|
||||
|
||||
/// Sender reports roughly once per second.
|
||||
func senderReports(now: Date = Date()) -> [[UInt8]] {
|
||||
guard let srtp, now >= reportsDue else { return [] }
|
||||
reportsDue = now.addingTimeInterval(1)
|
||||
var out: [[UInt8]] = []
|
||||
if let audio {
|
||||
out.append(srtp.protectRTCP(RTCPFeedback.senderReport(ssrc: audio.ssrc, rtpTimestamp: lastAudioTimestamp,
|
||||
packets: audio.packetCount, octets: audio.octetCount, now: now)))
|
||||
}
|
||||
if let video {
|
||||
out.append(srtp.protectRTCP(RTCPFeedback.senderReport(ssrc: video.ssrc, rtpTimestamp: lastVideoTimestamp,
|
||||
packets: video.packetCount, octets: video.octetCount, now: now)))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/// No consent for 30 s (RFC 7675), or never connected within 30 s, or DTLS ended.
|
||||
func isExpired(now: Date = Date()) -> Bool {
|
||||
switch dtls.state {
|
||||
case .closed, .failed: return true
|
||||
default: break
|
||||
}
|
||||
return now.timeIntervalSince(lastConsent ?? created) > 30
|
||||
}
|
||||
|
||||
private static func token(_ length: Int) -> String {
|
||||
let alphabet = Array("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
|
||||
return String((0..<length).map { _ in alphabet.randomElement()! })
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,72 @@
|
||||
import CoreVideo
|
||||
import Foundation
|
||||
import Testing
|
||||
@testable import Intercom
|
||||
|
||||
@Suite struct MediaTests {
|
||||
@Test func opusEncodesTwentyMillisecondPackets() throws {
|
||||
let encoder = try #require(TalkbackEncoder())
|
||||
let tone = (0..<4800).map { Float(sin(Double($0) * 2 * .pi * 440 / 48_000) * 0.3) }
|
||||
let packets = tone.withUnsafeBufferPointer { encoder.append($0) }
|
||||
#expect(packets.count == 5)
|
||||
#expect(packets.allSatisfy { !$0.0.isEmpty && $0.0.count < 400 })
|
||||
let steps = zip(packets.dropFirst(), packets).map { $0.1 &- $1.1 }
|
||||
#expect(steps.allSatisfy { $0 == 960 })
|
||||
}
|
||||
|
||||
@Test(.timeLimit(.minutes(1))) func previewStartsWithParameterSetsAndKeyFrame() async throws {
|
||||
let encoder = PreviewEncoder(width: 320, height: 180)
|
||||
let received = Received()
|
||||
encoder.onAccessUnit = { nalUnits, _ in received.add(nalUnits) }
|
||||
|
||||
var buffer: CVPixelBuffer?
|
||||
CVPixelBufferCreate(nil, 1280, 720, kCVPixelFormatType_32BGRA,
|
||||
[kCVPixelBufferIOSurfacePropertiesKey: [:]] as CFDictionary, &buffer)
|
||||
let frame = try #require(buffer)
|
||||
for _ in 0..<20 where received.first == nil {
|
||||
encoder.encode(frame)
|
||||
try await Task.sleep(for: .milliseconds(50))
|
||||
}
|
||||
let nalUnits = try #require(received.first)
|
||||
let types = nalUnits.map { $0[0] & 0x1F }
|
||||
#expect(types.prefix(2) == [7, 8]) // SPS, PPS
|
||||
#expect(types.contains(5)) // IDR
|
||||
// Constrained baseline: profile_idc 66, constraint_set1 flag.
|
||||
#expect(nalUnits[0][1] == 66)
|
||||
#expect(nalUnits[0][2] & 0x40 != 0)
|
||||
}
|
||||
|
||||
private final class Received: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var units: [[[UInt8]]] = []
|
||||
func add(_ nalUnits: [[UInt8]]) { lock.withLock { units.append(nalUnits) } }
|
||||
var first: [[UInt8]]? { lock.withLock { units.first } }
|
||||
}
|
||||
}
|
||||
|
||||
@Suite struct HTTPTests {
|
||||
@Test func parsesRequestAcrossChunks() throws {
|
||||
var parser = HTTPRequestParser()
|
||||
let raw = Array("POST /api/offer?x=1 HTTP/1.1\r\nHost: musubi\r\nContent-Length: 11\r\n\r\n{\"a\":\"bcd\"}".utf8)
|
||||
#expect(try parser.append(Array(raw[0..<30])) == nil)
|
||||
let request = try #require(try parser.append(Array(raw[30...])))
|
||||
#expect(request.method == "POST")
|
||||
#expect(request.path == "/api/offer")
|
||||
#expect(request.headers["host"] == "musubi")
|
||||
#expect(request.body == Array("{\"a\":\"bcd\"}".utf8))
|
||||
}
|
||||
|
||||
@Test func rejectsOversizedBody() {
|
||||
var parser = HTTPRequestParser()
|
||||
#expect(throws: HTTPRequestParser.ParseError.self) {
|
||||
try parser.append(Array("POST / HTTP/1.1\r\nContent-Length: 999999\r\n\r\n".utf8))
|
||||
}
|
||||
}
|
||||
|
||||
@Test func serializesResponse() {
|
||||
let text = String(decoding: HTTPResponse.text(404, "Nicht gefunden.").serialized, as: UTF8.self)
|
||||
#expect(text.hasPrefix("HTTP/1.1 404 Not Found\r\n"))
|
||||
#expect(text.contains("Content-Length: 15\r\n"))
|
||||
#expect(text.hasSuffix("\r\n\r\nNicht gefunden."))
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue