| 1 | import Foundation |
| 2 | import RastrilloApple |
| 3 | |
| 4 | @MainActor extension SheetModel { |
| 5 | func watch() async { |
| 6 | guard page != nil else { return } |
| 7 | let source = grid |
| 8 | // Paces reconnects: a server's Retry-After is a deadline no wake crosses; |
| 9 | // otherwise back off, cut short by a connectivity change, with no timer |
| 10 | // while the path is down. The backoff resets only after a stream that |
| 11 | // stayed open for 5 s, timed from onOpen. |
| 12 | let reconnect = NativeReconnect(connectivity: connectivity, now: reconnectClock) |
| 13 | while !Task.isCancelled && grid == source { |
| 14 | live = connectivity.isOffline(client.origin) ? "Offline" : "Connecting" |
| 15 | do { |
| 16 | // Subscribe before fetching the snapshot in onOpen; buffered deltas |
| 17 | // then reconcile by version once the initial fetch finishes. |
| 18 | let parser = SheetEventParser() |
| 19 | try await client.stream( |
| 20 | source + "/stream", |
| 21 | onOpen: { [weak self] in |
| 22 | await reconnect.opened() |
| 23 | await self?.streamOpened(source: source) |
| 24 | } |
| 25 | ) { [weak self] line in |
| 26 | let event = await parser.receive(line) |
| 27 | guard let event else { return } |
| 28 | await self?.received(event, source: source) |
| 29 | } |
| 30 | } catch is CancellationError { return } catch { |
| 31 | guard !Task.isCancelled, grid == source else { return } |
| 32 | reconnect.failed(error) |
| 33 | // The failure itself says whether this server was unreachable; the |
| 34 | // connectivity record of it arrives asynchronously. |
| 35 | var offline = connectivity.isOffline(client.origin) |
| 36 | if case NativeClientError.offline = error { offline = true } |
| 37 | live = offline ? "Offline" : "Reconnecting" |
| 38 | } |
| 39 | if live == "Access changed" || live == "Paused" { return } |
| 40 | if !connectivity.isPathSatisfied { live = "Offline" } |
| 41 | await reconnect.pause() |
| 42 | if Task.isCancelled { return } |
| 43 | } |
| 44 | } |
| 45 | private func streamOpened(source: String) async { |
| 46 | guard !Task.isCancelled, grid == source else { return } |
| 47 | await refresh() |
| 48 | guard !Task.isCancelled, grid == source else { return } |
| 49 | live = "Live" |
| 50 | } |
| 51 | func received(_ event: SheetEvent, source: String) async { |
| 52 | guard !Task.isCancelled, grid == source else { return } |
| 53 | switch event.name { |
| 54 | case "cells": |
| 55 | live = "Live" |
| 56 | if let response = try? await wire.decode(Data(event.data.utf8)) { |
| 57 | guard !Task.isCancelled, grid == source else { return } |
| 58 | await accept(response) |
| 59 | } else { |
| 60 | guard !Task.isCancelled, grid == source else { return } |
| 61 | await refresh() |
| 62 | } |
| 63 | case "resync": await refresh() |
| 64 | case "bye": |
| 65 | if event.data.contains("access-changed") { |
| 66 | live = "Access changed" |
| 67 | cells = [:] |
| 68 | styles = [:] |
| 69 | error = "Your access changed. Reopen the sheet to continue." |
| 70 | } else if event.data.contains("idle") { |
| 71 | live = "Paused" |
| 72 | } |
| 73 | default: live = "Live" |
| 74 | } |
| 75 | } |
| 76 | } |
| 77 | struct SheetEvent: Sendable { |
| 78 | let name: String |
| 79 | let data: String |
| 80 | } |
| 81 | actor SheetEventParser { |
| 82 | private var name = "" |
| 83 | private var data = "" |
| 84 | func receive(_ line: String) -> SheetEvent? { |
| 85 | if line.isEmpty { |
| 86 | let event = SheetEvent(name: name, data: data) |
| 87 | name = "" |
| 88 | data = "" |
| 89 | return event |
| 90 | } |
| 91 | if line.hasPrefix("event:") { |
| 92 | name = String(line.dropFirst(6)).trimmingCharacters(in: .whitespaces) |
| 93 | } |
| 94 | if line.hasPrefix("data:") { |
| 95 | if !data.isEmpty { data += "\n" } |
| 96 | data += line.dropFirst(5).trimmingCharacters(in: .whitespaces) |
| 97 | } |
| 98 | return nil |
| 99 | } |
| 100 | } |
| 101 | |
amadan