iOS 原生客户端：libwebrtc 数据面 + UX 改造 + 品牌/字体 + 全平台平均速度
- 原生数据面 libwebrtc（PLAN 决策 F）：新增 Engine/LibWebRtcEngine.swift（RTCPeerConnection，端口 engine.go/session.go 线协议 + 水位背压 + ack 完成 + 不冲突落盘）；NativeTransfer.swift 的 P2PEventsBridge 改 conform P2PEngineDelegate（去 gomobile）；EngineController 4 个 p2p RPC 改调新引擎（completion-based + resolveOnMain）。stasel/WebRTC M149 经 SPM（exactVersion 149.0.0）。环回端到端 20MB 逐字节测过（Tests/） - UX 改造（PARITY 残余项 1/2）：新增 SendComposeSheet（发送预览＝文件名 + 选设备，手动多选与分享共用）；分享直入预览 + scenePhase .active 兜底（修“分享后必须重启 app”）；重写 MessagesView（按设备会话 + 每设备未读红点 + 删单条 / 删整设备历史）；EngineController 加 conversations / unread 模型 - 品牌 logo：新增 Shared/Brand.swift（BrandMark / Wordmark / BrandLockup / BrandHero）+ Brand.xcassets/LogoMark；登录页 / 传输空态 / 设置头 / 控制中心控件接 logo；Theme 加 cdropAmber - 字体（对齐 web，仅主 app）：新增 BrandFonts.swift（Orbit Gothic + Maple Mono woff2 运行期 CTFontManager 注册 + UIKit 外观代理 + .oXxx / .maple 助手）；Fredoka SemiBold 字标；7 视图 .font sweep。字体不入库——.gitignore 排除 + just ios-fonts 构建前从 CDN / Google Fonts 拉取，见 Sources/Fonts/README.md - 分享引导：新增 Share/ShareGuideView（吉祥物卡 + “打开 Commilitia Drop”）；ShareViewController 健壮 staging（文件 / 数据双路）+ 响应链 IMP 唤起宿主 - 全平台平均速度：传输完成态显平均速度（总字节 / 总耗时）而非末次瞬时——web store/slices/transfer.ts + TransferRow.tsx + iOS RootView 卡片 / 详情；i18n 加 transfer.avgRate - 工程：project.yml 接 SPM WebRTC + CDropTests + Fredoka UIAppFonts + Share i18n；Justfile 加 ios-sim-build + ios-fonts / ios-fonts-fredoka 配方 - 文档：ios/PLAN.md（决策 F + §10 UX + §11 品牌/字体 + §12 平均速度）、ios/PARITY.md（字体分叉条目更新 + 2026-06-28 漂移行）
This commit is contained in:
@@ -30,6 +30,23 @@ struct MessageItem: Identifiable, Equatable, Codable
|
||||
let sentAt: Double
|
||||
}
|
||||
|
||||
// 会话摘要（消息按设备聚合，社交软件式）：每个对端设备一条，驱动消息 tab 的会话列表。由
|
||||
// conversations() 从 messages（取每设备最新一条）∪ 当前可达设备（在线 / 可唤醒，供发起新会话）合成。
|
||||
// reachable＝在线或可经 APNs 唤醒（离线 iOS）——决定 ConversationView 是否可发：不可达的离线设备
|
||||
// （离线桌面 / 浏览器）仅可查看历史、不可发（对齐 sendableDevices 语义）。
|
||||
struct ConversationItem: Identifiable, Equatable
|
||||
{
|
||||
let peerName: String
|
||||
let deviceID: String // 匹配到的当前设备 id（用于图标 / 判型）；无匹配（历史残留设备）为空
|
||||
let deviceType: String
|
||||
let online: Bool
|
||||
let reachable: Bool // 在线或可唤醒 → 可发；否则仅查看
|
||||
let lastText: String
|
||||
let lastAt: Double // 最近一条消息时间（epoch 秒）；无历史（仅可达）为 0，排在有历史之后
|
||||
let lastOutgoing: Bool
|
||||
var id: String { peerName }
|
||||
}
|
||||
|
||||
// 引擎过桥推来的传输记录（对应 web engine/main.ts 的 toWire 精简视图）。ice* 为选中候选
|
||||
// 对诊断（揭示 P2P 是否实为 TURN 中继，解释慢速）。同样手动容忍解析。
|
||||
struct TransferItem: Identifiable, Equatable, Codable
|
||||
@@ -68,6 +85,24 @@ struct LogEntry: Identifiable, Equatable
|
||||
let level: String // warn | error | info
|
||||
let message: String
|
||||
let at: Date
|
||||
// 从 "session <id> ..." 形式的诊断行解析出的传输会话 id（引擎诊断行 + pion 内部日志统一此格式），
|
||||
// 供日志界面按会话聚合 / 整体复制；非会话日志（SSE / 鉴权等）为 nil。
|
||||
let sessionId: String?
|
||||
|
||||
init(level: String, message: String, at: Date)
|
||||
{
|
||||
self.level = level
|
||||
self.message = message
|
||||
self.at = at
|
||||
self.sessionId = LogEntry.parseSession(message)
|
||||
}
|
||||
|
||||
static func parseSession(_ msg: String) -> String?
|
||||
{
|
||||
guard msg.hasPrefix("session ") else { return nil }
|
||||
let id = msg.dropFirst("session ".count).prefix { !$0.isWhitespace }
|
||||
return id.isEmpty ? nil : String(id)
|
||||
}
|
||||
}
|
||||
|
||||
// 待批准的他端登录请求（#8）：本机（已登录）扫到另一台设备显示的登录二维码（/link?r=&c=）后，经引擎
|
||||
@@ -117,10 +152,13 @@ final class EngineController: NSObject
|
||||
// 待确认的他端登录请求（#8）：扫码拿到信息后置位，驱动设备页弹批准确认；批准 / 拒绝 / 取消后清空。
|
||||
var pendingLoginRequest: LoginRequestItem?
|
||||
|
||||
// 未读红点（#5）：消息 / 传输各自的未读计数，对应 tab 激活时清零。activeTab 记录当前所在
|
||||
// 标签，使「正在该页时到达的事件」不计未读。seenIncomingTransfers 去重，避免同一传入传输在
|
||||
// 活跃→完成两阶段被重复计数。
|
||||
var unreadMessages = 0
|
||||
// 未读红点（#5）：传输未读对应 tab 激活时清零（activeTab 记录当前页，使「正在该页时到达的事件」不
|
||||
// 计未读）；seenIncomingTransfers 去重，避免同一传入传输在活跃→完成两阶段被重复计数。
|
||||
// 消息未读按设备分桶（unreadByPeer）：每设备各自未读计数（会话列表逐行红点），消息 tab 徽标＝总和
|
||||
// （unreadMessages 计算属性）；activeConversation 记录当前打开的会话，使「正在该会话时到达的消息」不计未读。
|
||||
var unreadByPeer: [String: Int] = [:]
|
||||
var activeConversation: String?
|
||||
var unreadMessages: Int { unreadByPeer.values.reduce(0, +) }
|
||||
var unreadTransfers = 0
|
||||
var activeTab = "transfer"
|
||||
private var seenIncomingTransfers: Set<String> = []
|
||||
@@ -155,6 +193,14 @@ final class EngineController: NSObject
|
||||
// 回取字节（下方 WKURLSchemeHandler 供给）。
|
||||
private var outgoing: [String: URL] = [:]
|
||||
|
||||
// 原生数据面引擎（libwebrtc，Safari 同款 RTCPeerConnection）：逃离 WKWebView 写死的 256KB SCTP
|
||||
// 接收窗 + 渲染器节流（见 desktop/NATIVE-TRANSFER.md §7）。取代早先 gomobile/pion——pion 的 Go raw
|
||||
// socket 不与 iOS Network framework 集成，真机确证不可靠。懒构造（首次 P2P RPC 时建），落盘＝
|
||||
// Documents（与 DownloadManager 同源，故收到的文件经「收到的文件」视图可见）。p2pEvents 强持引用
|
||||
// （引擎只弱引它），避免回调宿主被释放。libwebrtc 自管线程（信令 / 网络线程），无需宿主串行队列。
|
||||
private var transferEngine: LibWebRtcEngine?
|
||||
private var p2pEvents: P2PEventsBridge?
|
||||
|
||||
// 历史 / 消息限长（原生持久化，跨重启存活）。
|
||||
private static let historyCap = 50
|
||||
private static let messagesCap = 200
|
||||
@@ -276,7 +322,8 @@ final class EngineController: NSObject
|
||||
// 登出清空瞬时态：通知 / 日志 / 未读 / 已读传输集 / 待批准登录，不跨账号残留。
|
||||
notices = []
|
||||
logs = []
|
||||
unreadMessages = 0
|
||||
unreadByPeer = [:]
|
||||
activeConversation = nil
|
||||
unreadTransfers = 0
|
||||
seenIncomingTransfers = []
|
||||
activeTab = "transfer"
|
||||
@@ -307,7 +354,11 @@ final class EngineController: NSObject
|
||||
func clearHistory()
|
||||
{
|
||||
history.removeAll()
|
||||
RecordsStore.clear(recordKey("history"))
|
||||
// 落空数组到分桶键（而非删文件）：activateRecords 在分桶键缺失时回退读 legacy 未分桶文件，
|
||||
// 直接删文件会让旧记录重启复活（清空按钮失效的根因）；写 [] 使 load 返回空数组、阻断回退。
|
||||
// 顺手清掉 legacy 未分桶文件。
|
||||
persistHistory()
|
||||
RecordsStore.clear("history")
|
||||
}
|
||||
|
||||
// 删除一条消息（H）/ 清空全部（持久化同步）。
|
||||
@@ -320,7 +371,18 @@ final class EngineController: NSObject
|
||||
func clearMessages()
|
||||
{
|
||||
messages.removeAll()
|
||||
RecordsStore.clear(recordKey("messages"))
|
||||
// 同 clearHistory：落空数组阻断 legacy 回退（删文件会令清空后的消息重启复活），再清 legacy 文件。
|
||||
persistMessages()
|
||||
RecordsStore.clear("messages")
|
||||
}
|
||||
|
||||
// 删除与某设备的整段会话历史（按对端设备名）：清掉该设备的全部消息，避免废会话 / 历史残留堆积。
|
||||
// 仅删本地展示记录（持久化同步），不影响对方。删空后该会话从列表消失（除非该设备当前可达，
|
||||
// 仍作「可发起新会话」项保留）。
|
||||
func deleteConversation(_ peerName: String)
|
||||
{
|
||||
messages.removeAll { $0.peerName == peerName }
|
||||
persistMessages()
|
||||
}
|
||||
|
||||
// ---- 瞬时通知（#2） ----
|
||||
@@ -350,10 +412,28 @@ final class EngineController: NSObject
|
||||
func setActiveTab(_ tab: String)
|
||||
{
|
||||
activeTab = tab
|
||||
if tab == "messages" { unreadMessages = 0 }
|
||||
// 消息未读不在进入 tab 时清零（那只是看到会话列表、尚未读各设备消息）——改为打开某会话时清该
|
||||
// 设备未读（见 markConversationRead）。传输未读仍按 tab 激活清零。
|
||||
if tab == "transfer" { unreadTransfers = 0 }
|
||||
}
|
||||
|
||||
// 打开某设备会话（ConversationView.onAppear）：记录当前会话 + 清该设备未读；关闭（onDisappear）置 nil。
|
||||
func setActiveConversation(_ peerName: String?)
|
||||
{
|
||||
activeConversation = peerName
|
||||
}
|
||||
|
||||
func markConversationRead(_ peerName: String)
|
||||
{
|
||||
if unreadByPeer[peerName] != nil { unreadByPeer[peerName] = 0 }
|
||||
}
|
||||
|
||||
// 某设备的未读消息数（会话列表行红点用）。
|
||||
func unread(for peerName: String) -> Int
|
||||
{
|
||||
return unreadByPeer[peerName] ?? 0
|
||||
}
|
||||
|
||||
// noteIncomingTransfer：传入传输首次出现（活跃或完成）时计一次未读，按 sessionId 去重，且仅当
|
||||
// 不在传输页时计。本机发出的传输不计。
|
||||
private func noteIncomingTransfer(_ item: TransferItem)
|
||||
@@ -405,6 +485,57 @@ final class EngineController: NSObject
|
||||
return Self.isWakeable(dev) ? "\(dev.name) · \(t("home.device.wakeable"))" : dev.name
|
||||
}
|
||||
|
||||
// ---- 消息按设备聚合（社交软件式会话） ----
|
||||
|
||||
// 会话列表：每个对端设备一条。取「有消息历史的设备」∪「当前可达设备（在线 / 可唤醒，供发起新
|
||||
// 会话）」，匹配当前 presence 取设备型号 / 在线态。按最近消息时间降序（无历史的可达设备 lastAt=0
|
||||
// 排在最后），最近活跃的会话在最前。
|
||||
func conversations() -> [ConversationItem]
|
||||
{
|
||||
// 每个对端的最新一条消息（messages 最新在前，故首次见到即最新）。
|
||||
var latest: [String: MessageItem] = [:]
|
||||
for m in messages where latest[m.peerName] == nil { latest[m.peerName] = m }
|
||||
|
||||
let reachable = sendableDevices()
|
||||
let reachableNames = Set(reachable.map { $0.name })
|
||||
|
||||
// 候选对端：有历史的 ∪ 当前可达的（去重）。
|
||||
var peers = Set(latest.keys)
|
||||
peers.formUnion(reachableNames)
|
||||
peers.remove(deviceName) // 不与本机会话
|
||||
|
||||
var items: [ConversationItem] = []
|
||||
for peer in peers
|
||||
{
|
||||
let dev = devices.first { $0.name == peer }
|
||||
let last = latest[peer]
|
||||
items.append(ConversationItem(
|
||||
peerName: peer,
|
||||
deviceID: dev?.deviceID ?? "",
|
||||
deviceType: dev?.type ?? "",
|
||||
online: dev?.online ?? false,
|
||||
reachable: reachableNames.contains(peer),
|
||||
lastText: last?.text ?? "",
|
||||
lastAt: last?.sentAt ?? 0,
|
||||
lastOutgoing: last?.direction == "outgoing"))
|
||||
}
|
||||
// 最近消息在前；时间相同（多为均无历史）按名稳定排序。
|
||||
return items.sorted { a, b in a.lastAt != b.lastAt ? a.lastAt > b.lastAt : a.peerName < b.peerName }
|
||||
}
|
||||
|
||||
// 与某设备的消息线程（最旧→最新，供聊天视图自上而下渲染）。
|
||||
func thread(with peerName: String) -> [MessageItem]
|
||||
{
|
||||
return messages.filter { $0.peerName == peerName }.sorted { $0.sentAt < $1.sentAt }
|
||||
}
|
||||
|
||||
// 某对端当前是否可发消息：在线或可经 APNs 唤醒（＝在 sendableDevices 中）。不可达的离线设备
|
||||
// （离线桌面 / 浏览器）会话仅可查看历史、撰写栏禁用。
|
||||
func canMessage(_ peerName: String) -> Bool
|
||||
{
|
||||
return sendableDevices().contains { $0.name == peerName }
|
||||
}
|
||||
|
||||
// 剪贴板上行：读本机 UIPasteboard 文本交引擎上传云剪贴板（用户手势触发，前台读取合规）。
|
||||
func uploadClipboard()
|
||||
{
|
||||
@@ -617,6 +748,54 @@ final class EngineController: NSObject
|
||||
let data = (try handle.read(upToCount: count)) ?? Data()
|
||||
return data.base64EncodedString()
|
||||
}
|
||||
|
||||
// ---- 原生数据面（gomobile/pion）----
|
||||
|
||||
// 懒构造原生引擎：落盘目录＝Documents（与 DownloadManager 同源），事件经 P2PEventsBridge
|
||||
// 跳回主线程过桥给 JS。libwebrtc 工厂构造不会失败（连接级失败按会话经 RPC error 上报）。
|
||||
private func nativeTransferEngine() -> LibWebRtcEngine
|
||||
{
|
||||
if let e = transferEngine { return e }
|
||||
let events = P2PEventsBridge()
|
||||
events.controller = self
|
||||
let dir = FileManager.default.urls(for: .documentDirectory, in: .userDomainMask)[0].path
|
||||
let e = LibWebRtcEngine(downloadDir: dir, delegate: events)
|
||||
p2pEvents = events
|
||||
transferEngine = e
|
||||
return e
|
||||
}
|
||||
|
||||
// 把 cdrop-file://<id> 引用解析回暂存文件的沙盒绝对路径（Go 据此直接读盘发送）。安全作用域
|
||||
// 访问已在 stageOutgoingFile 开启、持续到 reset，故同进程的 Go 可直接 os.Open 该路径。无对应
|
||||
// 暂存项返回空串。
|
||||
private func resolveOutgoingPath(_ ref: String) -> String
|
||||
{
|
||||
guard let id = URL(string: ref)?.host, let url = outgoing[id] else { return "" }
|
||||
return url.path
|
||||
}
|
||||
|
||||
// resolveOnMain 把引擎完成回调（在 libwebrtc 信令线程触发，非主线程）跳回主线程 resolve 该 RPC。
|
||||
// nonisolated 使其可从任意线程调用；内部经 DispatchQueue.main + assumeIsolated 进 @MainActor 调
|
||||
// resolve（evaluateJavaScript 仅限主线程）。error≠nil → ok=false（该会话回退中继 / 报错）。
|
||||
nonisolated private func resolveOnMain(id: Int, error: Error?)
|
||||
{
|
||||
DispatchQueue.main.async
|
||||
{
|
||||
MainActor.assumeIsolated
|
||||
{
|
||||
self.resolve(id: id, ok: error == nil, value: error?.localizedDescription)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 原生引擎诊断日志入环形缓冲（#3 日志界面）。由 P2PEventsBridge.onLog 在主线程调用。
|
||||
func appendNativeLog(_ msg: String)
|
||||
{
|
||||
guard !msg.isEmpty else { return }
|
||||
lastEngineLog = msg
|
||||
logs.insert(LogEntry(level: "info", message: msg, at: Date()), at: 0)
|
||||
if logs.count > Self.logsCap { logs.removeLast(logs.count - Self.logsCap) }
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - JS → 原生（RPC + 通知）
|
||||
@@ -669,6 +848,40 @@ extension EngineController: WKScriptMessageHandler
|
||||
start: Self.intOf(payload["start"]),
|
||||
end: Self.intOf(payload["end"]))
|
||||
resolve(id: id, ok: true, value: b64)
|
||||
case "p2pStartOutgoing":
|
||||
// 发起原生发送：filePath 为 cdrop-file://<id>，解析回沙盒绝对路径交引擎直接读盘。引擎
|
||||
// 调用立即返回（libwebrtc 异步），completion 在 offer 已铸+发出后回，经 resolveOnMain 回交。
|
||||
let engine = nativeTransferEngine()
|
||||
let path = resolveOutgoingPath(payload["filePath"] as? String ?? "")
|
||||
guard !path.isEmpty
|
||||
else { resolve(id: id, ok: false, value: "no staged file"); return }
|
||||
let sid = payload["sessionId"] as? String ?? ""
|
||||
let peer = payload["peerName"] as? String ?? ""
|
||||
let ice = payload["iceServersJSON"] as? String ?? ""
|
||||
engine.startOutgoing(sessionId: sid, peerName: peer, filePath: path, iceServersJSON: ice)
|
||||
{ [weak self] err in self?.resolveOnMain(id: id, error: err) }
|
||||
case "p2pStartIncoming":
|
||||
// 预备原生接收：建 PeerConnection 等对端 offer，收到即直接写盘到 Documents。
|
||||
let engine = nativeTransferEngine()
|
||||
let sid = payload["sessionId"] as? String ?? ""
|
||||
let peer = payload["peerName"] as? String ?? ""
|
||||
let ice = payload["iceServersJSON"] as? String ?? ""
|
||||
engine.startIncoming(sessionId: sid, peerName: peer, iceServersJSON: ice)
|
||||
{ [weak self] err in self?.resolveOnMain(id: id, error: err) }
|
||||
case "p2pHandleSignal":
|
||||
// 把 JS hub 从 SSE 收到的入站信令喂给原生引擎（按对端设备名路由）。completion 在信令应用
|
||||
// 完成后回——JS 侧信令链据此按序投递（offer/answer 先于其后 trickle 候选，见 p2pIos.ts chain）。
|
||||
guard let engine = transferEngine
|
||||
else { resolve(id: id, ok: false, value: "no native session"); return }
|
||||
let from = payload["fromPeer"] as? String ?? ""
|
||||
let pj = payload["payloadJSON"] as? String ?? ""
|
||||
engine.handleSignal(fromPeer: from, payloadJSON: pj)
|
||||
{ [weak self] err in self?.resolveOnMain(id: id, error: err) }
|
||||
case "p2pCancel":
|
||||
// 取消 / 中继回退：拆除原生会话（静默，不回 OnState）。libwebrtc close 快，直接调，立即 resolve。
|
||||
let sid = payload["sessionId"] as? String ?? ""
|
||||
transferEngine?.cancel(sessionId: sid)
|
||||
resolve(id: id, ok: true, value: nil)
|
||||
default:
|
||||
resolve(id: id, ok: false, value: "unknown method: \(method)")
|
||||
}
|
||||
@@ -729,8 +942,12 @@ extension EngineController: WKScriptMessageHandler
|
||||
messages.removeLast(messages.count - Self.messagesCap)
|
||||
}
|
||||
persistMessages()
|
||||
// 新到的对端消息：不在消息页时计未读红点（#5）。本机发出的乐观回显（outgoing）不计。
|
||||
if isNew, item.direction == "incoming", activeTab != "messages" { unreadMessages += 1 }
|
||||
// 新到的对端消息：非「正在该设备会话中」即给该设备计未读（按设备分桶）。本机发出的
|
||||
// 乐观回显（outgoing）不计。
|
||||
if isNew, item.direction == "incoming", item.peerName != activeConversation
|
||||
{
|
||||
unreadByPeer[item.peerName, default: 0] += 1
|
||||
}
|
||||
}
|
||||
case "transferDone":
|
||||
// 完成项移出活跃、前插历史；按 sessionId 去重避免重复事件叠加。
|
||||
|
||||
@@ -0,0 +1,905 @@
|
||||
import Foundation
|
||||
import WebRTC
|
||||
|
||||
// LibWebRtcEngine —— iOS 原生数据面引擎，用 libwebrtc（Safari 同款 RTCPeerConnection）在原生进程内
|
||||
// 完成 P2P 文件收发。取代早先 gomobile/pion 路径：pion 的 Go raw socket 不与 iOS Network framework
|
||||
// 集成（WebKit/libwebrtc 才集成），真机确证架构性不可靠（见 desktop/NATIVE-TRANSFER.md §7）。
|
||||
//
|
||||
// 线协议与 desktop/engine（pion）及 web/src/features/transfer/p2p.ts 逐字节一致，故可与桌面 pion
|
||||
// 引擎、浏览器 / 旧 iOS 的 JS 引擎互通：DataChannel "cdrop-file"（ordered），控制帧 meta/done/ack 走
|
||||
// 文本帧、文件分片走二进制帧（64KB），信令 payload 形如 {type, sdp?, candidate?}。本类是 session.go +
|
||||
// engine.go 的忠实 Swift 端口——逻辑、水位、ack 语义、收尾顺序一一对应。
|
||||
//
|
||||
// 分工同桌面：本引擎只跑纯 WebRTC + 文件 I/O，一切 HTTP（出站信令 POST / 状态机 POST）留 JS
|
||||
// （见 web/src/features/transfer/p2pIos.ts）。事件经 P2PEngineDelegate 回灌——宿主即 EngineController
|
||||
// 经 cdropEngine 桥（P2PEventsBridge）转给无头 JS 引擎。
|
||||
//
|
||||
// 宿主契约（EngineController 的 4 个 p2p RPC + net/ios.ts 的 nativeP2P*）一字不改；只把 Swift 侧引擎
|
||||
// 从 gomobile EngineEngine 换成本类。翻 web 侧 IOS_NATIVE=true 即启用。
|
||||
|
||||
// MARK: - 事件回调（宿主实现，等价 desktop/engine 的 Events 接口）
|
||||
|
||||
protocol P2PEngineDelegate: AnyObject
|
||||
{
|
||||
// 已传字节：发送端＝接收端 ack 追平值（无 ack 时回退本地已交付估计），接收端＝已写盘字节。
|
||||
func engineDidProgress(sessionId: String, bytes: Int64)
|
||||
// 会话状态：connected / completed / failed / closed。
|
||||
func engineDidChangeState(sessionId: String, state: String)
|
||||
// 出站信令（offer/answer/ice）：payloadJSON 为信令 JSON 串，宿主 POST /api/hub/signal。
|
||||
func engineDidEmitSignal(sessionId: String, toPeer: String, payloadJSON: String)
|
||||
// 接收端落盘完成的绝对路径（沙盒 Documents）。
|
||||
func engineDidSave(sessionId: String, path: String)
|
||||
// 选中候选对（形如 "host/udp"），供调试面板展示实际连通路径。
|
||||
func engineDidSelectIcePair(sessionId: String, local: String, remote: String)
|
||||
// 诊断日志（候选 / ICE 态 / 选中对 / 发送停滞），形如 "session <id> ..."，供日志界面按会话聚合。
|
||||
func engineDidLog(_ line: String)
|
||||
}
|
||||
|
||||
// MARK: - 线协议常量（与 desktop/engine wire.go、web p2p.ts 一致）
|
||||
|
||||
private enum Wire
|
||||
{
|
||||
static let channelName = "cdrop-file"
|
||||
static let chunkSize = 64 * 1024
|
||||
static let highWatermark: UInt64 = 16 * 1024 * 1024
|
||||
static let lowWatermark: UInt64 = 4 * 1024 * 1024
|
||||
static let ackIntervalMs = 200
|
||||
static let progressThrottleNs: Int64 = 100 * 1_000_000 // ~10Hz，护宿主主线程（每片过桥 evaluateJavaScript）
|
||||
static let ackCompleteTimeout: TimeInterval = 30
|
||||
// ICE 候选预热池，与 web p2p.ts ICE_CANDIDATE_POOL 对齐。
|
||||
static let iceCandidatePool: Int32 = 4
|
||||
}
|
||||
|
||||
struct EngineError: LocalizedError
|
||||
{
|
||||
let msg: String
|
||||
init(_ m: String) { msg = m }
|
||||
var errorDescription: String? { msg }
|
||||
}
|
||||
|
||||
// MARK: - 引擎（持有所有活跃会话，按对端设备名索引，等价 engine.go 的 Engine）
|
||||
|
||||
final class LibWebRtcEngine
|
||||
{
|
||||
// 单次 SSL 初始化（libwebrtc 要求）。全局 let 惰性、线程安全地执行一次。
|
||||
private static let sslReady: Bool = RTCInitializeSSL()
|
||||
|
||||
private let factory: RTCPeerConnectionFactory
|
||||
// 内部可设：生产由 init 注入（EngineController 的 P2PEventsBridge）；测试可重定向到环回宿主。
|
||||
weak var delegate: P2PEngineDelegate?
|
||||
|
||||
private let lock = NSLock()
|
||||
private var byPeer: [String: P2PTransferSession] = [:]
|
||||
private var downloadDirValue: String
|
||||
|
||||
init(downloadDir: String, delegate: P2PEngineDelegate)
|
||||
{
|
||||
_ = LibWebRtcEngine.sslReady
|
||||
factory = RTCPeerConnectionFactory()
|
||||
downloadDirValue = downloadDir
|
||||
self.delegate = delegate
|
||||
}
|
||||
|
||||
var events: P2PEngineDelegate? { delegate }
|
||||
|
||||
func downloadDir() -> String
|
||||
{
|
||||
lock.lock(); defer { lock.unlock() }
|
||||
return downloadDirValue
|
||||
}
|
||||
|
||||
func setDownloadDir(_ dir: String)
|
||||
{
|
||||
lock.lock(); downloadDirValue = dir; lock.unlock()
|
||||
}
|
||||
|
||||
// StartOutgoing：建 PeerConnection + DataChannel，发 offer，通道就绪后流式发出 filePath。
|
||||
// filePath 为本机绝对路径（原生从 cdrop-file://<id> 解析得到）。completion 在 offer 已铸并
|
||||
// setLocalDescription 完成后回（或出错时回错误），对齐 JS 侧信令链 seed 语义（见 p2pIos.ts chain）。
|
||||
func startOutgoing(sessionId: String, peerName: String, filePath: String,
|
||||
iceServersJSON: String, completion: @escaping (Error?) -> Void)
|
||||
{
|
||||
let fm = FileManager.default
|
||||
var isDir: ObjCBool = false
|
||||
guard fm.fileExists(atPath: filePath, isDirectory: &isDir), !isDir.boolValue
|
||||
else { completion(EngineError("not a file: \(filePath)")); return }
|
||||
let size = ((try? fm.attributesOfItem(atPath: filePath))?[.size] as? NSNumber)?.int64Value ?? 0
|
||||
|
||||
let s = P2PTransferSession(engine: self, factory: factory,
|
||||
sessionId: sessionId, peerName: peerName, role: .sender)
|
||||
s.filePath = filePath
|
||||
s.fileSize = size
|
||||
do { try s.createPeerConnection(iceServersJSON) }
|
||||
catch { completion(error); return }
|
||||
put(peerName, s)
|
||||
do { try s.createSenderDataChannel() }
|
||||
catch { s.fail("create datachannel: \(error.localizedDescription)"); completion(error); return }
|
||||
s.createOfferAndEmit(completion: completion)
|
||||
}
|
||||
|
||||
// StartIncoming：建 PeerConnection 等对端 offer，收到 DataChannel 后按线协议落盘。须在对端
|
||||
// offer 到达前调用。completion 立即回（会话已登记，后续入站 offer/ice 可路由）。
|
||||
func startIncoming(sessionId: String, peerName: String,
|
||||
iceServersJSON: String, completion: @escaping (Error?) -> Void)
|
||||
{
|
||||
let s = P2PTransferSession(engine: self, factory: factory,
|
||||
sessionId: sessionId, peerName: peerName, role: .receiver)
|
||||
do { try s.createPeerConnection(iceServersJSON) }
|
||||
catch { completion(error); return }
|
||||
put(peerName, s)
|
||||
completion(nil)
|
||||
}
|
||||
|
||||
// HandleSignal：把宿主从 SSE 收到的入站信令喂给对应会话（按对端设备名路由）。completion 在信令
|
||||
// 应用完成后回——对 offer 是 answer 已铸并发出后，对 answer 是 setRemoteDescription 完成后，对
|
||||
// ice 是 addIceCandidate 完成后。这保证 JS 侧信令链按序投递（offer/answer 先于其后 trickle 候选）。
|
||||
func handleSignal(fromPeer: String, payloadJSON: String, completion: @escaping (Error?) -> Void)
|
||||
{
|
||||
lock.lock(); let s = byPeer[fromPeer]; lock.unlock()
|
||||
guard let s
|
||||
else { completion(EngineError("no active session for peer \(fromPeer)")); return }
|
||||
s.handleSignal(payloadJSON, completion: completion)
|
||||
}
|
||||
|
||||
// Cancel：拆除指定会话（取消 / 中继回退时由宿主调用）。静默拆除，不回 OnState。
|
||||
func cancel(sessionId: String)
|
||||
{
|
||||
lock.lock()
|
||||
var target: P2PTransferSession?
|
||||
for (_, s) in byPeer where s.sessionId == sessionId { target = s; break }
|
||||
lock.unlock()
|
||||
target?.cleanup()
|
||||
}
|
||||
|
||||
fileprivate func put(_ peer: String, _ s: P2PTransferSession)
|
||||
{
|
||||
lock.lock()
|
||||
let old = byPeer[peer]
|
||||
byPeer[peer] = s
|
||||
lock.unlock()
|
||||
// 同一对端若已有会话，先拆旧的（沿用 web / 桌面按对端覆盖的语义）。异步拆，避免在 RPC / 主线程
|
||||
// 上同步 close（对齐 engine.go 的 `go old.cleanup()`）。
|
||||
if let old, old !== s { DispatchQueue.global(qos: .utility).async { old.cleanup() } }
|
||||
}
|
||||
|
||||
fileprivate func remove(_ peer: String, _ s: P2PTransferSession)
|
||||
{
|
||||
lock.lock()
|
||||
if byPeer[peer] === s { byPeer.removeValue(forKey: peer) }
|
||||
lock.unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - 单条传输会话（pc + dc + 角色相关收发循环，等价 session.go 的 session）
|
||||
|
||||
final class P2PTransferSession: NSObject
|
||||
{
|
||||
enum Role { case sender, receiver }
|
||||
|
||||
let sessionId: String
|
||||
let peerName: String
|
||||
let role: Role
|
||||
private weak var engine: LibWebRtcEngine?
|
||||
private let factory: RTCPeerConnectionFactory
|
||||
|
||||
private var pc: RTCPeerConnection?
|
||||
private var dc: RTCDataChannel?
|
||||
|
||||
// 发送端
|
||||
var filePath = ""
|
||||
var fileSize: Int64 = 0
|
||||
|
||||
// 计数器（跨线程：发送循环 / pion 回调 / dc 委托）。cLock 守护。
|
||||
private let cLock = NSLock()
|
||||
private var bytesSent: Int64 = 0 // 已塞进本地 SCTP 缓冲的字节
|
||||
private var receivedBytes: Int64 = 0 // 接收端已写盘字节
|
||||
private var ackedBytes: Int64 = 0 // 接收端经 ack 回传的已收字节
|
||||
private var lastEmitNs: Int64 = 0 // 进度节流时间戳
|
||||
|
||||
// 接收端落盘。outLock 守护 outHandle 的读/写/关——writeChunk 在 libwebrtc 信令线程跑，cleanup 可能
|
||||
// 在主线程（JS p2pCancel / 中继回退 / SSE 终态）并发关句柄；无锁则 ARC retain/release 与 close/write
|
||||
// 同体竞争 → 可能崩溃（Go 的 *os.File 有内建 fdMutex 故安全，Swift FileHandle 无此保证）。
|
||||
private var outHandle: FileHandle?
|
||||
private var outName = ""
|
||||
private var tmpPath = ""
|
||||
private let outLock = NSLock()
|
||||
|
||||
// 发送流 / 背压
|
||||
private let streamQueue: DispatchQueue
|
||||
private let bufSem = DispatchSemaphore(value: 0) // didChangeBufferedAmount / ack 唤醒等待循环
|
||||
private var ackTimer: DispatchSourceTimer?
|
||||
private var senderStarted = false
|
||||
|
||||
// 生命周期
|
||||
private let stateLock = NSLock()
|
||||
private var state = ""
|
||||
private var doneFlag = false
|
||||
private var cleanedUp = false
|
||||
|
||||
init(engine: LibWebRtcEngine, factory: RTCPeerConnectionFactory,
|
||||
sessionId: String, peerName: String, role: Role)
|
||||
{
|
||||
self.engine = engine
|
||||
self.factory = factory
|
||||
self.sessionId = sessionId
|
||||
self.peerName = peerName
|
||||
self.role = role
|
||||
streamQueue = DispatchQueue(label: "net.commilitia.cdrop.p2p.\(sessionId)")
|
||||
super.init()
|
||||
}
|
||||
|
||||
// MARK: 建连
|
||||
|
||||
func createPeerConnection(_ iceServersJSON: String) throws
|
||||
{
|
||||
let config = RTCConfiguration()
|
||||
config.iceServers = Self.parseICEServers(iceServersJSON)
|
||||
config.sdpSemantics = .unifiedPlan
|
||||
config.continualGatheringPolicy = .gatherOnce
|
||||
config.iceCandidatePoolSize = Wire.iceCandidatePool
|
||||
// 默认网络处理（不禁链路本地、不强 IPv4、不限接口）：libwebrtc 与 iOS Network framework 原生
|
||||
// 集成，同内网经 host↔host 直连。绝不在此加 iOS 定向过滤——曾把 IPv4-only 漏进桌面共享引擎致
|
||||
// 桌面↔iOS-WebKit 的 IPv6 host 对被裁、退中继（重大回归）。
|
||||
let constraints = RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil)
|
||||
guard let pc = factory.peerConnection(with: config, constraints: constraints, delegate: self)
|
||||
else { throw EngineError("failed to create peer connection") }
|
||||
self.pc = pc
|
||||
}
|
||||
|
||||
func createSenderDataChannel() throws
|
||||
{
|
||||
guard let pc else { throw EngineError("no pc") }
|
||||
let dcConfig = RTCDataChannelConfiguration()
|
||||
dcConfig.isOrdered = true
|
||||
guard let dc = pc.dataChannel(forLabel: Wire.channelName, configuration: dcConfig)
|
||||
else { throw EngineError("failed to create data channel") }
|
||||
dc.delegate = self
|
||||
self.dc = dc
|
||||
}
|
||||
|
||||
func createOfferAndEmit(completion: @escaping (Error?) -> Void)
|
||||
{
|
||||
guard let pc else { completion(EngineError("no pc")); return }
|
||||
let constraints = RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil)
|
||||
pc.offer(for: constraints)
|
||||
{ [weak self] sdp, err in
|
||||
guard let self else { completion(EngineError("session gone")); return }
|
||||
if let err { self.fail("create offer: \(err.localizedDescription)"); completion(err); return }
|
||||
guard let sdp else { self.fail("create offer: nil"); completion(EngineError("nil offer")); return }
|
||||
pc.setLocalDescription(sdp)
|
||||
{ [weak self] err2 in
|
||||
guard let self else { completion(EngineError("session gone")); return }
|
||||
if let err2
|
||||
{
|
||||
self.fail("set local (offer): \(err2.localizedDescription)"); completion(err2); return
|
||||
}
|
||||
self.emitSignal(sdp: sdp, candidate: nil)
|
||||
completion(nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: 入站信令
|
||||
|
||||
func handleSignal(_ payloadJSON: String, completion: @escaping (Error?) -> Void)
|
||||
{
|
||||
guard let pc else { completion(EngineError("no pc")); return }
|
||||
guard let data = payloadJSON.data(using: .utf8),
|
||||
let obj = (try? JSONSerialization.jsonObject(with: data)) as? [String: Any],
|
||||
let type = obj["type"] as? String
|
||||
else { completion(EngineError("decode signal")); return }
|
||||
log("signal in: \(type)")
|
||||
|
||||
switch type
|
||||
{
|
||||
case "offer":
|
||||
guard let sdpObj = obj["sdp"] as? [String: Any], let sdpStr = sdpObj["sdp"] as? String
|
||||
else { completion(EngineError("offer missing sdp")); return }
|
||||
let desc = RTCSessionDescription(type: .offer, sdp: sdpStr)
|
||||
pc.setRemoteDescription(desc)
|
||||
{ [weak self] err in
|
||||
guard let self else { completion(EngineError("gone")); return }
|
||||
if let err { completion(err); return }
|
||||
let constraints = RTCMediaConstraints(mandatoryConstraints: nil, optionalConstraints: nil)
|
||||
pc.answer(for: constraints)
|
||||
{ [weak self] ans, aerr in
|
||||
guard let self else { completion(EngineError("gone")); return }
|
||||
if let aerr { completion(aerr); return }
|
||||
guard let ans else { completion(EngineError("nil answer")); return }
|
||||
pc.setLocalDescription(ans)
|
||||
{ [weak self] lerr in
|
||||
guard let self else { completion(EngineError("gone")); return }
|
||||
if let lerr { completion(lerr); return }
|
||||
self.emitSignal(sdp: ans, candidate: nil)
|
||||
completion(nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
case "answer":
|
||||
guard let sdpObj = obj["sdp"] as? [String: Any], let sdpStr = sdpObj["sdp"] as? String
|
||||
else { completion(EngineError("answer missing sdp")); return }
|
||||
let desc = RTCSessionDescription(type: .answer, sdp: sdpStr)
|
||||
pc.setRemoteDescription(desc) { err in completion(err) }
|
||||
case "ice":
|
||||
// 空候选 / end-of-candidates：忽略（pion 同样跳过 nil candidate）。
|
||||
guard let candObj = obj["candidate"] as? [String: Any],
|
||||
let candStr = candObj["candidate"] as? String, !candStr.isEmpty
|
||||
else { completion(nil); return }
|
||||
let mid = candObj["sdpMid"] as? String
|
||||
let mline = (candObj["sdpMLineIndex"] as? NSNumber)?.int32Value ?? 0
|
||||
let cand = RTCIceCandidate(sdp: candStr, sdpMLineIndex: mline, sdpMid: mid)
|
||||
pc.add(cand) { err in completion(err) }
|
||||
default:
|
||||
completion(nil)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: 出站信令
|
||||
|
||||
// 与 session.go emitSignal 同形：有 candidate 即 ice，否则取 sdp 自身类型（offer/answer）。
|
||||
// JSON 形状与 web p2p.ts 的 SignalPayload / pion 的 signalPayload 逐字段对齐（键序无关，两侧皆 parse）。
|
||||
private func emitSignal(sdp: RTCSessionDescription?, candidate: RTCIceCandidate?)
|
||||
{
|
||||
var payload: [String: Any] = [:]
|
||||
if let candidate
|
||||
{
|
||||
payload["type"] = "ice"
|
||||
// RTCIceCandidate（ObjC）不暴露 usernameFragment，故 candidate JSON 仅含 candidate/sdpMid/
|
||||
// sdpMLineIndex。ufrag 由 m-line 的 ice-ufrag 决定（offer/answer 已带），pion 与浏览器的
|
||||
// addIceCandidate 对缺省 usernameFragment 皆容忍，互通不受影响。
|
||||
var c: [String: Any] = [
|
||||
"candidate": candidate.sdp,
|
||||
"sdpMLineIndex": Int(candidate.sdpMLineIndex),
|
||||
]
|
||||
c["sdpMid"] = candidate.sdpMid ?? NSNull()
|
||||
payload["candidate"] = c
|
||||
}
|
||||
else if let sdp
|
||||
{
|
||||
let t = Self.sdpTypeString(sdp.type)
|
||||
payload["type"] = t
|
||||
payload["sdp"] = [ "type": t, "sdp": sdp.sdp ]
|
||||
}
|
||||
else { return }
|
||||
|
||||
guard let data = try? JSONSerialization.data(withJSONObject: payload),
|
||||
let json = String(data: data, encoding: .utf8)
|
||||
else { return }
|
||||
engine?.events?.engineDidEmitSignal(sessionId: sessionId, toPeer: peerName, payloadJSON: json)
|
||||
}
|
||||
|
||||
// MARK: 发送端
|
||||
|
||||
private func startSenderStream()
|
||||
{
|
||||
cLock.lock()
|
||||
if senderStarted { cLock.unlock(); return }
|
||||
senderStarted = true
|
||||
cLock.unlock()
|
||||
streamQueue.async { [weak self] in self?.streamFile() }
|
||||
}
|
||||
|
||||
// streamFile 把文件按 64KB 分片经数据通道发出，水位回压与桌面 pion / web 引擎一致：缓冲将越
|
||||
// HIGH 前等抽干到 LOW。发完 done 帧后先抽干本地缓冲、再等接收端 ack 追平总量才宣告 completed。
|
||||
private func streamFile()
|
||||
{
|
||||
guard let dc else { fail("no data channel"); return }
|
||||
let fh: FileHandle
|
||||
do { fh = try FileHandle(forReadingFrom: URL(fileURLWithPath: filePath)) }
|
||||
catch { fail("open \(filePath): \(error.localizedDescription)"); return }
|
||||
defer { try? fh.close() }
|
||||
|
||||
if !sendText([ "type": "meta", "name": (filePath as NSString).lastPathComponent, "size": fileSize ])
|
||||
{
|
||||
fail("send meta"); return
|
||||
}
|
||||
|
||||
while true
|
||||
{
|
||||
if isDone() { return }
|
||||
let chunk: Data?
|
||||
do { chunk = try fh.read(upToCount: Wire.chunkSize) }
|
||||
catch { fail("read \(filePath): \(error.localizedDescription)"); return }
|
||||
guard let chunk, !chunk.isEmpty else { break } // EOF
|
||||
|
||||
if dc.bufferedAmount + UInt64(chunk.count) > Wire.highWatermark
|
||||
{
|
||||
waitBufferLow(Wire.lowWatermark)
|
||||
}
|
||||
if isDone() { return }
|
||||
if !dc.sendData(RTCDataBuffer(data: chunk, isBinary: true))
|
||||
{
|
||||
fail("send chunk failed (channel closed?)"); return
|
||||
}
|
||||
addSent(Int64(chunk.count))
|
||||
emitProgress()
|
||||
}
|
||||
|
||||
_ = sendText([ "type": "done" ])
|
||||
waitDrained()
|
||||
waitAck(fileSize)
|
||||
emitProgressNow()
|
||||
setState("completed")
|
||||
}
|
||||
|
||||
@discardableResult
|
||||
private func sendText(_ dict: [String: Any]) -> Bool
|
||||
{
|
||||
guard let dc, let data = try? JSONSerialization.data(withJSONObject: dict) else { return false }
|
||||
return dc.sendData(RTCDataBuffer(data: data, isBinary: false))
|
||||
}
|
||||
|
||||
// 等本地缓冲落到 threshold 以下：靠 didChangeBufferedAmount 唤醒 bufSem，兼 50ms 轮询兜底。
|
||||
private func waitBufferLow(_ threshold: UInt64)
|
||||
{
|
||||
let start = Date()
|
||||
var warned = false
|
||||
while true
|
||||
{
|
||||
guard let dc else { return }
|
||||
if dc.bufferedAmount <= threshold { return }
|
||||
if isDone() { return }
|
||||
_ = bufSem.wait(timeout: .now() + 0.05)
|
||||
if !warned, Date().timeIntervalSince(start) > 5
|
||||
{
|
||||
warned = true
|
||||
log("send stalled >5s: buffered=\(dc.bufferedAmount) sent=\(getSent()) acked=\(getAcked())")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 等本地缓冲彻底抽干（done 帧也已离开本端 SCTP）。
|
||||
private func waitDrained()
|
||||
{
|
||||
while true
|
||||
{
|
||||
guard let dc else { return }
|
||||
if dc.bufferedAmount == 0 { return }
|
||||
if isDone() { return }
|
||||
_ = bufSem.wait(timeout: .now() + 0.02)
|
||||
}
|
||||
}
|
||||
|
||||
// 等接收端 ack 追平 total（确认已全收）。超时兜底；无任何 ack 则不空等（对齐 session.go waitAck）。
|
||||
private func waitAck(_ total: Int64)
|
||||
{
|
||||
if getAcked() == 0 { return }
|
||||
let deadline = Date().addingTimeInterval(Wire.ackCompleteTimeout)
|
||||
while getAcked() < total
|
||||
{
|
||||
if Date() > deadline { return }
|
||||
if isDone() { return }
|
||||
_ = bufSem.wait(timeout: .now() + 0.05)
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: 接收端
|
||||
|
||||
private func bindReceiverDataChannel(_ channel: RTCDataChannel)
|
||||
{
|
||||
dc = channel
|
||||
channel.delegate = self
|
||||
startAckLoop()
|
||||
}
|
||||
|
||||
// 每 200ms 把最新已收字节回传发送端（节流，避免每片都回与下行争用通道）。
|
||||
private func startAckLoop()
|
||||
{
|
||||
let timer = DispatchSource.makeTimerSource(queue: streamQueue)
|
||||
timer.schedule(deadline: .now() + .milliseconds(Wire.ackIntervalMs),
|
||||
repeating: .milliseconds(Wire.ackIntervalMs))
|
||||
var lastSent: Int64 = -1
|
||||
timer.setEventHandler
|
||||
{ [weak self] in
|
||||
guard let self else { return }
|
||||
let r = self.getReceived()
|
||||
if r != lastSent { self.sendAck(r); lastSent = r }
|
||||
}
|
||||
// 若 cleanup 已先发生（极少见：didOpen 与 cancel 同刻），不再挂表，避免泄漏一个活计时器。
|
||||
stateLock.lock()
|
||||
if cleanedUp { stateLock.unlock(); timer.cancel(); return }
|
||||
ackTimer = timer
|
||||
stateLock.unlock()
|
||||
timer.resume()
|
||||
}
|
||||
|
||||
private func sendAck(_ bytes: Int64)
|
||||
{
|
||||
guard let dc, dc.readyState == .open,
|
||||
let data = try? JSONSerialization.data(withJSONObject: [ "type": "ack", "bytes": bytes ])
|
||||
else { return }
|
||||
_ = dc.sendData(RTCDataBuffer(data: data, isBinary: false))
|
||||
}
|
||||
|
||||
// 文本控制帧：发送端处理 ack，接收端处理 meta / done。角色守护避免误处理对端不该发的帧。
|
||||
private func handleControl(_ data: Data)
|
||||
{
|
||||
guard let obj = (try? JSONSerialization.jsonObject(with: data)) as? [String: Any],
|
||||
let type = obj["type"] as? String
|
||||
else { return }
|
||||
switch type
|
||||
{
|
||||
case "meta":
|
||||
if role == .receiver { openOutput(name: obj["name"] as? String ?? "download") }
|
||||
case "done":
|
||||
if role == .receiver { finalizeReceive() }
|
||||
case "ack":
|
||||
if role == .sender, let b = (obj["bytes"] as? NSNumber)?.int64Value
|
||||
{
|
||||
setAcked(b)
|
||||
bufSem.signal() // 唤醒 waitAck / waitDrained
|
||||
emitProgress()
|
||||
}
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
private func openOutput(name: String)
|
||||
{
|
||||
outName = name
|
||||
let dir = engine?.downloadDir() ?? NSTemporaryDirectory()
|
||||
let tmp = (dir as NSString).appendingPathComponent(".\(sessionId).part")
|
||||
tmpPath = tmp
|
||||
FileManager.default.createFile(atPath: tmp, contents: nil)
|
||||
let handle = FileHandle(forWritingAtPath: tmp)
|
||||
outLock.lock(); outHandle = handle; outLock.unlock()
|
||||
if handle == nil { fail("create \(tmp)") }
|
||||
}
|
||||
|
||||
// 直接写盘——无 base64 桥、无 OPFS、无 WebView 内存。dc 委托串行投递，写入天然有序。
|
||||
private func writeChunk(_ data: Data)
|
||||
{
|
||||
// 持 outLock 整段写：与 cleanup 的关句柄互斥，杜绝写到已关闭 / 已释放的句柄（见 outHandle 注释）。
|
||||
outLock.lock()
|
||||
if let h = outHandle
|
||||
{
|
||||
do { try h.write(contentsOf: data) }
|
||||
catch { outLock.unlock(); fail("write chunk: \(error.localizedDescription)"); return }
|
||||
}
|
||||
outLock.unlock()
|
||||
_ = addReceived(Int64(data.count))
|
||||
emitProgress()
|
||||
}
|
||||
|
||||
// 收到 done＝所有分片已到（ordered），receivedBytes 已等于总量。回最终 ack、关文件、改名落到
|
||||
// 不冲突路径，报告路径与 completed。命名避开 NSObject 已弃用的 finalize()（否则被判为 override）。
|
||||
private func finalizeReceive()
|
||||
{
|
||||
sendAck(getReceived())
|
||||
outLock.lock(); let h = outHandle; outHandle = nil; outLock.unlock()
|
||||
if let h
|
||||
{
|
||||
try? h.close()
|
||||
let dir = engine?.downloadDir() ?? NSTemporaryDirectory()
|
||||
let final = Self.nonCollidingPath(dir: dir, name: outName)
|
||||
do { try FileManager.default.moveItem(atPath: tmpPath, toPath: final) }
|
||||
catch { fail("rename to \(final): \(error.localizedDescription)"); return }
|
||||
engine?.events?.engineDidSave(sessionId: sessionId, path: final)
|
||||
}
|
||||
emitProgressNow()
|
||||
setState("completed")
|
||||
}
|
||||
|
||||
// MARK: 进度 / 状态 / 收尾
|
||||
|
||||
// 当前应上报的已传字节：发送端优先用接收端 ack 追平值（无 ack 时回退本地已交付估计），
|
||||
// 接收端用已写盘字节（对齐 session.go progressBytes）。
|
||||
private func progressBytes() -> Int64
|
||||
{
|
||||
if role == .sender
|
||||
{
|
||||
let a = getAcked()
|
||||
if a > 0 { return min(fileSize, a) }
|
||||
let buffered = Int64(dc?.bufferedAmount ?? 0)
|
||||
let b = getSent() - buffered
|
||||
return b < 0 ? 0 : b
|
||||
}
|
||||
return getReceived()
|
||||
}
|
||||
|
||||
// 节流上报（~10Hz）：高吞吐下每片一回调把宿主主线程打满（每片过桥 evaluateJavaScript）。
|
||||
private func emitProgress()
|
||||
{
|
||||
let now = Int64(DispatchTime.now().uptimeNanoseconds)
|
||||
cLock.lock()
|
||||
if now - lastEmitNs < Wire.progressThrottleNs { cLock.unlock(); return }
|
||||
lastEmitNs = now
|
||||
cLock.unlock()
|
||||
engine?.events?.engineDidProgress(sessionId: sessionId, bytes: progressBytes())
|
||||
}
|
||||
|
||||
// 无视节流强发一次：收尾确保最终字节数到达宿主（否则末次进度被节流吞掉、文案停在 99%）。
|
||||
private func emitProgressNow()
|
||||
{
|
||||
cLock.lock(); lastEmitNs = Int64(DispatchTime.now().uptimeNanoseconds); cLock.unlock()
|
||||
engine?.events?.engineDidProgress(sessionId: sessionId, bytes: progressBytes())
|
||||
}
|
||||
|
||||
private func setState(_ st: String)
|
||||
{
|
||||
stateLock.lock()
|
||||
// 终态不可被覆盖：completed/failed 之后即便连接随后 closed/failed 也不回退。
|
||||
if state == st || state == "completed" || state == "failed"
|
||||
{
|
||||
stateLock.unlock(); return
|
||||
}
|
||||
state = st
|
||||
stateLock.unlock()
|
||||
|
||||
engine?.events?.engineDidChangeState(sessionId: sessionId, state: st)
|
||||
switch st
|
||||
{
|
||||
case "failed", "closed":
|
||||
cleanup()
|
||||
case "completed":
|
||||
// 不立即拆连接：接收端刚发最终 ack；高吞吐下反向 ack 可能未达发送端，此刻关通道会让发送端
|
||||
// 误判失败。留连接，ackLoop 续发；60s 兜底防泄漏。真正拆除交宿主 SSE 终态驱动的 cancel。
|
||||
streamQueue.asyncAfter(deadline: .now() + 60) { [weak self] in self?.cleanup() }
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
func fail(_ msg: String)
|
||||
{
|
||||
log("fail: \(msg)")
|
||||
setState("failed")
|
||||
}
|
||||
|
||||
func cleanup()
|
||||
{
|
||||
stateLock.lock()
|
||||
if cleanedUp { stateLock.unlock(); return }
|
||||
cleanedUp = true
|
||||
doneFlag = true
|
||||
let timer = ackTimer // 与 startAckLoop 的赋值经 stateLock 互斥，避免 ackTimer ARC 竞争
|
||||
ackTimer = nil
|
||||
stateLock.unlock()
|
||||
|
||||
bufSem.signal() // 唤醒任何在等缓冲 / ack 的发送循环
|
||||
timer?.cancel()
|
||||
// outHandle 经 outLock 取出再关：与 writeChunk 互斥（writeChunk 持锁期间 cleanup 在此阻塞，
|
||||
// 待其写完再关），杜绝并发 close/write 与 ARC 竞争。
|
||||
outLock.lock(); let h = outHandle; outHandle = nil; outLock.unlock()
|
||||
if let h { try? h.close() }
|
||||
dc?.close()
|
||||
pc?.close()
|
||||
engine?.remove(peerName, self)
|
||||
}
|
||||
|
||||
// MARK: 计数器（cLock 守护）
|
||||
|
||||
private func addSent(_ n: Int64) { cLock.lock(); bytesSent += n; cLock.unlock() }
|
||||
private func getSent() -> Int64 { cLock.lock(); defer { cLock.unlock() }; return bytesSent }
|
||||
private func addReceived(_ n: Int64) -> Int64
|
||||
{
|
||||
cLock.lock(); receivedBytes += n; let r = receivedBytes; cLock.unlock(); return r
|
||||
}
|
||||
private func getReceived() -> Int64 { cLock.lock(); defer { cLock.unlock() }; return receivedBytes }
|
||||
private func setAcked(_ v: Int64) { cLock.lock(); if v > ackedBytes { ackedBytes = v }; cLock.unlock() }
|
||||
private func getAcked() -> Int64 { cLock.lock(); defer { cLock.unlock() }; return ackedBytes }
|
||||
private func isDone() -> Bool { stateLock.lock(); defer { stateLock.unlock() }; return doneFlag }
|
||||
|
||||
// MARK: 选中候选对（连接建立后异步取，best-effort，仅供调试面板）
|
||||
|
||||
private func logSelectedPair()
|
||||
{
|
||||
// 不用 streamQueue：发送端的 streamFile 会长期占用它（背压等待中），用全局队列才能在传输期间
|
||||
// 实时取到选中候选对（调试面板用），而非等传输结束。
|
||||
DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + 0.3)
|
||||
{ [weak self] in self?.querySelectedPair(attempt: 0) }
|
||||
}
|
||||
|
||||
private func querySelectedPair(attempt: Int)
|
||||
{
|
||||
guard attempt < 12, !isDone(), let pc else { return }
|
||||
pc.statistics
|
||||
{ [weak self] report in
|
||||
guard let self else { return }
|
||||
if let pair = Self.extractSelectedPair(report)
|
||||
{
|
||||
self.log("ice pair: local=\(pair.0) remote=\(pair.1)")
|
||||
self.engine?.events?.engineDidSelectIcePair(sessionId: self.sessionId,
|
||||
local: pair.0, remote: pair.1)
|
||||
}
|
||||
else
|
||||
{
|
||||
DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + 0.3)
|
||||
{ [weak self] in self?.querySelectedPair(attempt: attempt + 1) }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: 静态助手
|
||||
|
||||
private static func parseICEServers(_ raw: String) -> [RTCIceServer]
|
||||
{
|
||||
let trimmed = raw.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
guard !trimmed.isEmpty, let data = trimmed.data(using: .utf8),
|
||||
let arr = (try? JSONSerialization.jsonObject(with: data)) as? [[String: Any]]
|
||||
else { return [] }
|
||||
var out: [RTCIceServer] = []
|
||||
for s in arr
|
||||
{
|
||||
var urls: [String] = []
|
||||
if let one = s["urls"] as? String { urls = [one] }
|
||||
else if let many = s["urls"] as? [String] { urls = many }
|
||||
if urls.isEmpty { continue }
|
||||
let username = s["username"] as? String
|
||||
let credential = s["credential"] as? String
|
||||
if let username, let credential, !username.isEmpty
|
||||
{
|
||||
out.append(RTCIceServer(urlStrings: urls, username: username, credential: credential))
|
||||
}
|
||||
else
|
||||
{
|
||||
out.append(RTCIceServer(urlStrings: urls))
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// 返回 dir 下不与现有文件冲突的目标路径：name、name (1)、name (2)……（对齐 wire.go nonCollidingPath）。
|
||||
private static func nonCollidingPath(dir: String, name: String) -> String
|
||||
{
|
||||
let fm = FileManager.default
|
||||
let base = (dir as NSString).appendingPathComponent(name)
|
||||
if !fm.fileExists(atPath: base) { return base }
|
||||
let ext = (name as NSString).pathExtension
|
||||
let stem = (name as NSString).deletingPathExtension
|
||||
var i = 1
|
||||
while true
|
||||
{
|
||||
let candidate = ext.isEmpty ? "\(stem) (\(i))" : "\(stem) (\(i)).\(ext)"
|
||||
let p = (dir as NSString).appendingPathComponent(candidate)
|
||||
if !fm.fileExists(atPath: p) { return p }
|
||||
i += 1
|
||||
}
|
||||
}
|
||||
|
||||
private static func extractSelectedPair(_ report: RTCStatisticsReport) -> (String, String)?
|
||||
{
|
||||
let stats = report.statistics
|
||||
var localId: String?
|
||||
var remoteId: String?
|
||||
for (_, s) in stats where s.type == "candidate-pair"
|
||||
{
|
||||
let v = s.values
|
||||
let succeeded = (v["state"] as? String) == "succeeded"
|
||||
let nominated = (v["nominated"] as? NSNumber)?.boolValue ?? false
|
||||
if succeeded, nominated
|
||||
{
|
||||
localId = v["localCandidateId"] as? String
|
||||
remoteId = v["remoteCandidateId"] as? String
|
||||
break
|
||||
}
|
||||
}
|
||||
guard let localId, let remoteId else { return nil }
|
||||
func label(_ id: String) -> String
|
||||
{
|
||||
guard let c = stats[id] else { return "?" }
|
||||
let t = (c.values["candidateType"] as? String) ?? "?"
|
||||
let proto = (c.values["protocol"] as? String) ?? "?"
|
||||
return "\(t)/\(proto)"
|
||||
}
|
||||
return (label(localId), label(remoteId))
|
||||
}
|
||||
|
||||
private static func sdpTypeString(_ t: RTCSdpType) -> String
|
||||
{
|
||||
switch t
|
||||
{
|
||||
case .offer: return "offer"
|
||||
case .answer: return "answer"
|
||||
case .prAnswer: return "pranswer"
|
||||
case .rollback: return "rollback"
|
||||
@unknown default: return "offer"
|
||||
}
|
||||
}
|
||||
|
||||
private static func iceConnString(_ s: RTCIceConnectionState) -> String
|
||||
{
|
||||
switch s
|
||||
{
|
||||
case .new: return "new"
|
||||
case .checking: return "checking"
|
||||
case .connected: return "connected"
|
||||
case .completed: return "completed"
|
||||
case .failed: return "failed"
|
||||
case .disconnected: return "disconnected"
|
||||
case .closed: return "closed"
|
||||
case .count: return "count"
|
||||
@unknown default: return "?"
|
||||
}
|
||||
}
|
||||
|
||||
// 从候选 SDP 串提取 "type/proto"（仅日志，best-effort）：
|
||||
// candidate:<foundation> <component> <proto> <priority> <ip> <port> typ <type> ...
|
||||
private static func candTypeString(_ sdp: String) -> String
|
||||
{
|
||||
let parts = sdp.split(separator: " ").map(String.init)
|
||||
var typ = "?"
|
||||
var proto = "?"
|
||||
if parts.count > 2 { proto = parts[2].lowercased() }
|
||||
if let i = parts.firstIndex(of: "typ"), i + 1 < parts.count { typ = parts[i + 1] }
|
||||
return "\(typ)/\(proto)"
|
||||
}
|
||||
|
||||
private func log(_ msg: String)
|
||||
{
|
||||
engine?.events?.engineDidLog("session \(sessionId) \(msg)")
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - RTCPeerConnectionDelegate（回调在 libwebrtc 信令线程，非主线程）
|
||||
|
||||
extension P2PTransferSession: RTCPeerConnectionDelegate
|
||||
{
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didChange stateChanged: RTCSignalingState) {}
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didAdd stream: RTCMediaStream) {}
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didRemove stream: RTCMediaStream) {}
|
||||
func peerConnectionShouldNegotiate(_ peerConnection: RTCPeerConnection) {}
|
||||
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCIceConnectionState)
|
||||
{
|
||||
log("ice: \(Self.iceConnString(newState))")
|
||||
}
|
||||
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCIceGatheringState)
|
||||
{
|
||||
if newState == .complete { log("local gathering done") }
|
||||
}
|
||||
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didGenerate candidate: RTCIceCandidate)
|
||||
{
|
||||
log("cand local: \(Self.candTypeString(candidate.sdp))")
|
||||
emitSignal(sdp: nil, candidate: candidate)
|
||||
}
|
||||
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didRemove candidates: [RTCIceCandidate]) {}
|
||||
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didOpen dataChannel: RTCDataChannel)
|
||||
{
|
||||
if role == .receiver { bindReceiverDataChannel(dataChannel) }
|
||||
}
|
||||
|
||||
// 组合 ice+dtls 连接态（等价 session.go OnConnectionStateChange）：驱动 connected/failed/closed。
|
||||
func peerConnection(_ peerConnection: RTCPeerConnection, didChange newState: RTCPeerConnectionState)
|
||||
{
|
||||
switch newState
|
||||
{
|
||||
case .connected:
|
||||
setState("connected")
|
||||
logSelectedPair()
|
||||
case .failed:
|
||||
fail("peerconnection failed")
|
||||
case .closed:
|
||||
setState("closed")
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// MARK: - RTCDataChannelDelegate（回调在 libwebrtc 信令线程）
|
||||
|
||||
extension P2PTransferSession: RTCDataChannelDelegate
|
||||
{
|
||||
func dataChannelDidChangeState(_ dataChannel: RTCDataChannel)
|
||||
{
|
||||
// 发送端：通道就绪即开始 streamFile（等价 session.go dc.OnOpen → go streamFile）。
|
||||
if role == .sender, dataChannel.readyState == .open { startSenderStream() }
|
||||
}
|
||||
|
||||
func dataChannel(_ dataChannel: RTCDataChannel, didReceiveMessageWith buffer: RTCDataBuffer)
|
||||
{
|
||||
if buffer.isBinary { writeChunk(buffer.data) }
|
||||
else { handleControl(buffer.data) }
|
||||
}
|
||||
|
||||
func dataChannel(_ dataChannel: RTCDataChannel, didChangeBufferedAmount amount: UInt64)
|
||||
{
|
||||
if amount <= Wire.lowWatermark { bufSem.signal() }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
import Foundation
|
||||
|
||||
// P2PEventsBridge：libwebrtc 原生数据面引擎（LibWebRtcEngine）的事件宿主。引擎在 libwebrtc 信令
|
||||
// 线程（非主线程）触发这些回调，每条都用 DispatchQueue.main.async 跳回主线程再交 EngineController
|
||||
// 过桥给无头 JS 引擎（evaluateJavaScript 仅限主线程）。
|
||||
//
|
||||
// 用 DispatchQueue.main.async 而非 Task { @MainActor }：前者对同源提交保持 FIFO，使
|
||||
// connected → completed 等有序状态不被乱序处理（Task 无顺序保证，completed 抢在 connected 前
|
||||
// 处理会把终态回退）。assumeIsolated 把主队列闭包桥进 EngineController 的 @MainActor 隔离域。
|
||||
//
|
||||
// 与桌面 desktop/app.go 的 transferEvents 等价——桌面经 Wails EventsEmit，iOS 经 cdropEngine 桥。
|
||||
// 事件契约见 LibWebRtcEngine.swift 的 P2PEngineDelegate；JS 侧消费见
|
||||
// web/src/features/transfer/p2pIos.ts（事件名 / payload 形状与桌面 NativeP2PHandlers 同形）。
|
||||
final class P2PEventsBridge: P2PEngineDelegate
|
||||
{
|
||||
weak var controller: EngineController?
|
||||
|
||||
// 已传字节（发送端＝接收端 ack 追平、接收端＝已写盘）。
|
||||
func engineDidProgress(sessionId: String, bytes: Int64)
|
||||
{
|
||||
emit("p2pProgress", [ "sessionId": sessionId, "bytes": bytes ])
|
||||
}
|
||||
|
||||
// 会话状态：connected / completed / failed / closed。
|
||||
func engineDidChangeState(sessionId: String, state: String)
|
||||
{
|
||||
emit("p2pState", [ "sessionId": sessionId, "state": state ])
|
||||
}
|
||||
|
||||
// 出站信令（offer/answer/ice）：payloadJSON 为信令 JSON 串，JS 侧 POST /api/hub/signal 前
|
||||
// 解析（payload 透传字符串再 JSON.parse，与桌面 p2pNative onSignal 同形）。
|
||||
func engineDidEmitSignal(sessionId: String, toPeer: String, payloadJSON: String)
|
||||
{
|
||||
emit("p2pSignal", [
|
||||
"sessionId": sessionId,
|
||||
"to": toPeer,
|
||||
"payload": payloadJSON,
|
||||
])
|
||||
}
|
||||
|
||||
// 接收端落盘完成的绝对路径（沙盒 Documents）。落盘提示由原生 transferDone 链路负责，JS 侧
|
||||
// p2pIos onSaved 当前为 no-op；事件仍过桥以备用。
|
||||
func engineDidSave(sessionId: String, path: String)
|
||||
{
|
||||
emit("p2pSaved", [ "sessionId": sessionId, "path": path ])
|
||||
}
|
||||
|
||||
// 选中候选对（形如 "host/udp"）：填进传输记录的 iceStats 供调试面板展示实际连通路径。
|
||||
func engineDidSelectIcePair(sessionId: String, local: String, remote: String)
|
||||
{
|
||||
emit("p2pIcePair", [
|
||||
"sessionId": sessionId,
|
||||
"local": local,
|
||||
"remote": remote,
|
||||
])
|
||||
}
|
||||
|
||||
// 引擎诊断日志（候选 / ICE 态 / 选中对 / 发送停滞）：直接进原生日志环形缓冲（#3 日志界面），
|
||||
// 不过 JS——诊断由原生侧呈现，便于真机现场排查同内网 mDNS/ICE 时间线。
|
||||
func engineDidLog(_ line: String)
|
||||
{
|
||||
let msg = line
|
||||
DispatchQueue.main.async { [weak controller] in
|
||||
MainActor.assumeIsolated { controller?.appendNativeLog(msg) }
|
||||
}
|
||||
}
|
||||
|
||||
private func emit(_ name: String, _ payload: [String: Any])
|
||||
{
|
||||
DispatchQueue.main.async { [weak controller] in
|
||||
MainActor.assumeIsolated { controller?.sendCommand(name, payload: payload) }
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user