aboutsummaryrefslogtreecommitdiff
path: root/Hotline/Library
diff options
context:
space:
mode:
Diffstat (limited to 'Hotline/Library')
-rw-r--r--Hotline/Library/HotlinePanel.swift6
-rw-r--r--Hotline/Library/NetSocket/FileProgress.swift58
-rw-r--r--Hotline/Library/NetSocket/NetSocketNew.swift (renamed from Hotline/Library/NetSocketNew.swift)1201
-rw-r--r--Hotline/Library/NetSocket/TransferRateEstimator.swift135
-rw-r--r--Hotline/Library/QuickLookPreviewView.swift22
-rw-r--r--Hotline/Library/URLAdditions.swift29
6 files changed, 773 insertions, 678 deletions
diff --git a/Hotline/Library/HotlinePanel.swift b/Hotline/Library/HotlinePanel.swift
index 191e89a..d7d8284 100644
--- a/Hotline/Library/HotlinePanel.swift
+++ b/Hotline/Library/HotlinePanel.swift
@@ -8,12 +8,12 @@ class HotlinePanel: NSPanel {
super.init(contentRect: NSRect(x: 0, y: 0, width: HOTLINE_PANEL_SIZE.width, height: HOTLINE_PANEL_SIZE.height), styleMask: [.nonactivatingPanel, .titled, .closable, .utilityWindow, .fullSizeContentView], backing: .buffered, defer: false)
// Make sure that the panel is in front of almost all other windows
- self.isFloatingPanel = false
+ self.isFloatingPanel = true
self.level = .floating
self.hidesOnDeactivate = true
self.animationBehavior = .utilityWindow
- // Allow the panel to appear in a fullscreen space
+ // Allow the panelto appear in a fullscreen space
// self.collectionBehavior.insert(.fullScreenAuxiliary)
self.collectionBehavior.insert(.canJoinAllSpaces)
self.collectionBehavior.insert(.ignoresCycle)
@@ -23,7 +23,7 @@ class HotlinePanel: NSPanel {
// Don't delete panel state when it's closed.
self.isReleasedWhenClosed = false
- self.standardWindowButton(.closeButton)?.isHidden = true
+ self.standardWindowButton(.closeButton)?.isHidden = false
self.standardWindowButton(.zoomButton)?.isHidden = true
self.standardWindowButton(.miniaturizeButton)?.isHidden = true
diff --git a/Hotline/Library/NetSocket/FileProgress.swift b/Hotline/Library/NetSocket/FileProgress.swift
new file mode 100644
index 0000000..c086af7
--- /dev/null
+++ b/Hotline/Library/NetSocket/FileProgress.swift
@@ -0,0 +1,58 @@
+// NetSocketProgress
+// Dustin Mierau • @mierau
+// MIT License
+
+import Foundation
+
+public extension NetSocketNew {
+
+ /// Progress information for file uploads/downloads
+ struct FileProgress: Sendable {
+ /// Number of bytes sent/received so far
+ public let sent: Int
+ /// Total file size (may be nil if unknown)
+ public let total: Int?
+ /// Smoothed transfer rate in bytes per second (EMA), if enough samples collected
+ public let bytesPerSecond: Double?
+ /// Estimated time remaining (seconds) based on smoothed rate, if available
+ public let estimatedTimeRemaining: TimeInterval?
+
+ public init(sent: Int, total: Int?, bytesPerSecond: Double? = nil, estimatedTimeRemaining: TimeInterval? = nil) {
+ self.sent = sent
+ self.total = total
+ self.bytesPerSecond = bytesPerSecond
+ self.estimatedTimeRemaining = estimatedTimeRemaining
+ }
+
+ /// Format transfer speed in human-readable format
+ ///
+ /// Automatically selects appropriate unit (B/sec, KB/sec, MB/sec, GB/sec)
+ /// based on the magnitude of the speed.
+ ///
+ /// - Returns: Formatted string like "45KB/sec", "5B/sec", "12.5MB/sec", or nil if speed unavailable
+ ///
+ /// Example:
+ /// ```swift
+ /// if let speedString = progress.formattedSpeed {
+ /// print(speedString) // "2.5MB/sec"
+ /// }
+ /// ```
+ public var formattedSpeed: String? {
+ guard let bytesPerSecond = bytesPerSecond, bytesPerSecond > 0 else { return nil }
+
+ let kb = 1024.0
+ let mb = kb * 1024.0
+ let gb = mb * 1024.0
+
+ if bytesPerSecond >= gb {
+ return String(format: "%.1fGB/sec", bytesPerSecond / gb)
+ } else if bytesPerSecond >= mb {
+ return String(format: "%.1fMB/sec", bytesPerSecond / mb)
+ } else if bytesPerSecond >= kb {
+ return String(format: "%.0fKB/sec", bytesPerSecond / kb)
+ } else {
+ return String(format: "%.0fB/sec", bytesPerSecond)
+ }
+ }
+ }
+}
diff --git a/Hotline/Library/NetSocketNew.swift b/Hotline/Library/NetSocket/NetSocketNew.swift
index 8873ee6..7b24b37 100644
--- a/Hotline/Library/NetSocketNew.swift
+++ b/Hotline/Library/NetSocket/NetSocketNew.swift
@@ -1,12 +1,9 @@
-
-// NetSocketNew.swift
-// Created by Dustin Mierau • @mierau
+// NetSocketNew.swift
+// Dustin Mierau • @mierau
import Foundation
import Network
-// MARK: - Endianness and Framing
-
/// Byte order for multi-byte integer values in binary protocols
public enum Endian {
/// Big-endian (network byte order, most significant byte first)
@@ -15,31 +12,6 @@ public enum Endian {
case little
}
-/// Length prefix types for framing variable-length data (strings, arrays, binary blobs)
-///
-/// Used to encode the size of the following data as a fixed-width integer.
-/// Each case can specify its own endianness.
-public enum LengthPrefix {
- /// 1-byte length prefix (0-255)
- case u8
- /// 2-byte length prefix (0-65,535)
- case u16(Endian = .big)
- /// 4-byte length prefix (0-4,294,967,295)
- case u32(Endian = .big)
- /// 8-byte length prefix (0-2^64-1)
- case u64(Endian = .big)
-
- /// Number of bytes used by this length prefix
- var byteCount: Int {
- switch self {
- case .u8: return 1
- case .u16: return 2
- case .u32: return 4
- case .u64: return 8
- }
- }
-}
-
/// Delimiter patterns for text-based protocols
public enum Delimiter {
/// Custom single byte delimiter
@@ -121,10 +93,8 @@ public enum NetSocketError: Error, CustomStringConvertible, Sendable {
/// NetSocketNew provides:
/// - Async connection management
/// - Automatic receive buffering with memory compaction
-/// - Length-prefixed framing for messages
/// - Type-safe reading/writing of integers, strings, and custom types
/// - File upload/download with progress tracking
-/// - Flexible encoder/decoder support (JSON, binary, etc.)
///
/// Example usage:
/// ```swift
@@ -139,46 +109,40 @@ public actor NetSocketNew {
public var receiveChunk: Int = 64 * 1024
/// Maximum bytes to buffer before disconnecting (default: 8 MB)
public var maxBufferBytes: Int = 8 * 1024 * 1024
- /// Maximum size for a single framed message (default: 4 MB)
- public var maxFrameBytes: Int = 4 * 1024 * 1024
public init() {}
}
-
+
// Connection + state
private let connection: NWConnection
private let queue = DispatchQueue(label: "NetSocket.NWConnection")
private var ready = false
private var isClosed = false
+ private let connectionID: String // For logging
// Buffer with compaction
private var buffer = Data()
private var head = 0 // start of unread bytes
- private let cfg: Config
+ private let config: Config
// Waiters for data/ready
private var dataWaiters: [CheckedContinuation<Void, Error>] = []
private var readyWaiters: [CheckedContinuation<Void, Error>] = []
- // Codable hooks - stored as closures for flexibility with any encoder/decoder
- private var encodeValue: @Sendable (any Encodable) throws -> Data = { value in
- let encoder = JSONEncoder()
- encoder.dateEncodingStrategy = .iso8601
- return try encoder.encode(value)
- }
-
- private var decodeValue: @Sendable (Data, any Decodable.Type) throws -> any Decodable = { data, type in
- let decoder = JSONDecoder()
- decoder.dateDecodingStrategy = .iso8601
- return try decoder.decode(type, from: data)
- }
-
- // MARK: Init/Connect
+ // MARK: Init
private init(connection: NWConnection, config: Config) {
self.connection = connection
- self.cfg = config
+ self.config = config
+ // Create a human-readable connection ID for logging
+ if case .hostPort(host: let h, port: let p) = connection.endpoint {
+ self.connectionID = "\(h):\(p)"
+ } else {
+ self.connectionID = "unknown"
+ }
}
+ // MARK: Connect
+
/// Connect to a remote host and return a ready socket
///
/// This method establishes a TCP connection using Network framework types and waits until
@@ -210,163 +174,7 @@ public actor NetSocketNew {
guard let nwPort = NWEndpoint.Port(rawValue: port) else {
throw NetSocketError.invalidPort
}
- return try await connect(host: .name(host, nil), port: nwPort, tls: tls, config: config)
- }
-
- /// Inject custom encoding/decoding logic (supports any encoder/decoder: JSON, CBOR, MessagePack, etc.)
- ///
- /// Example with JSONEncoder:
- /// ```
- /// let encoder = JSONEncoder()
- /// socket.useCoders(
- /// encode: { try encoder.encode($0) },
- /// decode: { data, type in try decoder.decode(type, from: data) }
- /// )
- /// ```
- ///
- /// Example with other encoders (pseudocode):
- /// ```
- /// let cbor = CBOREncoder()
- /// socket.useCoders(
- /// encode: { try cbor.encode($0) },
- /// decode: { data, type in try CBORDecoder().decode(type, from: data) }
- /// )
- /// ```
- public func useCoders(
- encode: @escaping @Sendable (any Encodable) throws -> Data,
- decode: @escaping @Sendable (Data, any Decodable.Type) throws -> any Decodable
- ) {
- self.encodeValue = encode
- self.decodeValue = decode
- }
-
- /// Convenience method to configure JSON encoding/decoding
- ///
- /// Sets up the socket to use the provided JSON encoder/decoder for `send()` and `receive()` calls.
- ///
- /// - Parameters:
- /// - encoder: A configured `JSONEncoder`
- /// - decoder: A configured `JSONDecoder`
- public func useJSONCoders(encoder: JSONEncoder, decoder: JSONDecoder) {
- self.encodeValue = { try encoder.encode($0) }
- self.decodeValue = { data, type in try decoder.decode(type, from: data) }
- }
-
- private func start() async throws {
- self.connection.stateUpdateHandler = { state in
- Task { [weak self] in
- guard let self else { return }
- switch state {
- case .ready:
- await self.setReady()
- await self.resumeReadyWaiters(with: .success(()))
- case .failed(let error):
- await self.failAllWaiters(NetSocketError.failed(underlying: error))
- await self.setClosed()
- case .waiting(let error):
- // bubble as transient failure for awaiters; reconnect logic could live here
- await self.resumeReadyWaiters(with: .failure(NetSocketError.failed(underlying: error)))
- case .cancelled:
- await self.failAllWaiters(NetSocketError.closed)
- await self.setClosed()
- default:
- break
- }
- }
- }
-
- // Kick off receive loop after .start
- self.connection.start(queue: queue)
- try await self.waitUntilReady()
- self.startReceiveLoop()
- }
-
- private func waitUntilReady() async throws {
- guard !ready else { return }
- try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
- readyWaiters.append(cont)
- }
- }
-
- private func resumeReadyWaiters(with result: Result<Void, Error>) {
- let waiters = readyWaiters
- readyWaiters.removeAll()
- for w in waiters {
- switch result {
- case .success: w.resume()
- case .failure(let e): w.resume(throwing: e)
- }
- }
- }
-
- private func failAllWaiters(_ error: Error) {
- resumeReadyWaiters(with: .failure(error))
- let waiters = dataWaiters
- dataWaiters.removeAll()
- for w in waiters { w.resume(throwing: error) }
- }
-
- private func setReady() {
- ready = true
- }
-
- private func setClosed() {
- isClosed = true
- }
-
- // MARK: Receive loop (runs on DispatchQueue, hops into actor)
-
- private nonisolated func startReceiveLoop() {
- func loop(_ connection: NWConnection, chunk: Int, owner: NetSocketNew) {
- print("NetSocketNew: Calling connection.receive() to request more data...")
- connection.receive(minimumIncompleteLength: 1, maximumLength: chunk) { data, _, isComplete, error in
- print("NetSocketNew: Receive callback - data: \(data?.count ?? 0) bytes, isComplete: \(isComplete), error: \(String(describing: error))")
- if let error {
- Task { await owner.handleReceiveError(error) }
- return
- }
- if let data, !data.isEmpty {
- Task { await owner.append(data) }
- }
- if isComplete {
- Task { await owner.handleEOF() }
- return
- }
- loop(connection, chunk: chunk, owner: owner)
- }
- }
- loop(connection, chunk: cfg.receiveChunk, owner: self)
- }
-
- private func handleReceiveError(_ error: Error) {
- isClosed = true
- failAllWaiters(NetSocketError.failed(underlying: error))
- }
-
- private func handleEOF() {
- isClosed = true
- let waiters = dataWaiters
- dataWaiters.removeAll()
- for w in waiters { w.resume() } // wake so readers can observe closure
- }
-
- private func append(_ data: Data) {
- print("NetSocketNew: Received \(data.count) bytes from network, buffer now has \(buffer.count - head + data.count) available")
- buffer.append(data)
- if buffer.count - head > cfg.maxBufferBytes {
- // Hard stop: drop connection rather than OOM'ing.
- isClosed = true
- connection.cancel()
- failAllWaiters(NetSocketError.framingExceeded(max: cfg.maxBufferBytes))
- return
- }
- resumeDataWaiters()
- }
-
- private func resumeDataWaiters() {
- let waiters = dataWaiters
- dataWaiters.removeAll()
- for w in waiters { w.resume() }
+ return try await self.connect(host: .name(host, nil), port: nwPort, tls: tls, config: config)
}
// MARK: Close
@@ -401,7 +209,7 @@ public actor NetSocketNew {
resumeReadyWaiters(with: .failure(NetSocketError.closed))
}
- // MARK: Send (async)
+ // MARK: Send Data
/// Write raw data to the socket
///
@@ -409,12 +217,13 @@ public actor NetSocketNew {
///
/// - Parameter data: Raw bytes to send
/// - Throws: `NetSocketError` if connection is not ready or send fails
- public func write(_ data: Data) async throws {
+ @discardableResult
+ public func write(_ data: Data) async throws -> Int {
try await ensureReady()
- try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
+ return try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Int, Error>) in
connection.send(content: data, completion: .contentProcessed { error in
if let error { cont.resume(throwing: NetSocketError.failed(underlying: error)) }
- else { cont.resume() }
+ else { cont.resume(returning: data.count) }
})
}
}
@@ -425,7 +234,8 @@ public actor NetSocketNew {
/// - value: The integer value to write
/// - endian: Byte order (default: big-endian)
/// - Throws: `NetSocketError` if write fails
- public func write<T: FixedWidthInteger>(_ value: T, endian: Endian = .big) async throws {
+ @discardableResult
+ public func write<T: FixedWidthInteger>(_ value: T, endian: Endian = .big) async throws -> Int {
var v = value
switch endian {
case .big: v = T(bigEndian: value)
@@ -437,155 +247,96 @@ public actor NetSocketNew {
Data(bytes: $0, count: size)
}
try await write(bytes)
+ return bytes.count
}
/// Write a boolean as a single byte (0 or 1)
/// - Parameter value: Boolean value
- public func write(_ value: Bool) async throws {
- try await write(value ? UInt8(0x01) : UInt8(0x00))
+ @discardableResult
+ public func write(_ value: Bool) async throws -> Int {
+ return try await write(UInt8(value ? 0x01 : 0x00))
}
/// Write a Float as its IEEE 754 bit pattern
/// - Parameters:
/// - value: Float value
/// - endian: Byte order (default: big-endian)
- public func write(_ value: Float, endian: Endian = .big) async throws {
- try await write(value.bitPattern, endian: endian)
+ @discardableResult
+ public func write(_ value: Float, endian: Endian = .big) async throws -> Int {
+ return try await write(value.bitPattern, endian: endian)
}
/// Write a Double as its IEEE 754 bit pattern
/// - Parameters:
/// - value: Double value
/// - endian: Byte order (default: big-endian)
- public func write(_ value: Double, endian: Endian = .big) async throws {
- try await write(value.bitPattern, endian: endian)
+ @discardableResult
+ public func write(_ value: Double, endian: Endian = .big) async throws -> Int {
+ return try await write(value.bitPattern, endian: endian)
}
/// Write a string to the socket, optionally length-prefixed
///
/// - Parameters:
/// - string: String to write
- /// - prefix: Optional length prefix (if provided, string is sent as a framed message)
/// - encoding: Text encoding (default: UTF-8)
+ /// - allowLossyConversion: Allow lossy encoding if necessary (default: false)
/// - Throws: `NetSocketError` if encoding fails or write fails
- public func write(_ string: String, prefix: LengthPrefix? = nil, encoding: String.Encoding = .utf8) async throws {
- guard let data = string.data(using: encoding) else {
+ @discardableResult
+ public func write(_ string: String, encoding: String.Encoding = .utf8, allowLossyConversion: Bool = false) async throws -> Int {
+ guard let data = string.data(using: encoding, allowLossyConversion: allowLossyConversion) else {
throw NetSocketError.encodeFailed(NSError(domain: "StringEncoding", code: -1))
}
- if let prefix { try await sendFrame(data, prefix: prefix) }
- else { try await write(data) }
+ return try await write(data)
}
- // MARK: Frames & Codable
+ // MARK: Receive Data
- /// Send a length-prefixed frame
- ///
- /// Writes the payload size as a fixed-width integer, followed by the payload bytes.
- ///
- /// - Parameters:
- /// - payload: Data to send
- /// - prefix: Length prefix type (default: u32 big-endian)
- /// - Throws: `NetSocketError.framingExceeded` if payload is too large for prefix type
- public func sendFrame(_ payload: Data, prefix: LengthPrefix = .u32()) async throws {
- // Ensure frame payload does not exceed max frame length.
- switch prefix {
- case .u8 where payload.count > Int(UInt8.max):
- throw NetSocketError.framingExceeded(max: Int(UInt8.max))
- case .u16 where payload.count > Int(UInt16.max):
- throw NetSocketError.framingExceeded(max: Int(UInt16.max))
- case .u32 where payload.count > Int(UInt32.max):
- throw NetSocketError.framingExceeded(max: Int(UInt32.max))
- default:
- break
- }
-
- if payload.count > cfg.maxFrameBytes { throw NetSocketError.framingExceeded(max: cfg.maxFrameBytes) }
- var header = Data()
- switch prefix {
- case .u8: header.append(UInt8(payload.count))
- case .u16(let e):
- try header.appendInteger(UInt16(payload.count), endian: e)
- case .u32(let e):
- try header.appendInteger(UInt32(payload.count), endian: e)
- case .u64(let e):
- try header.appendInteger(UInt64(payload.count), endian: e)
- }
- try await write(header + payload)
- }
-
- /// Receive a length-prefixed frame
- ///
- /// Reads a length prefix, then reads exactly that many bytes. Waits for data to arrive if needed.
- ///
- /// - Parameter prefix: Length prefix type (default: u32 big-endian)
- /// - Returns: The frame payload
- /// - Throws: `NetSocketError.framingExceeded` if frame size exceeds maximum
- public func receiveFrame(prefix: LengthPrefix = .u32()) async throws -> Data {
- let length: Int
- switch prefix {
- case .u8:
- let v: UInt8 = try await read(UInt8.self)
- length = Int(v)
- case .u16(let e):
- let v: UInt16 = try await read(UInt16.self, endian: e)
- length = Int(v)
- case .u32(let e):
- let v: UInt32 = try await read(UInt32.self, endian: e)
- length = Int(v)
- case .u64(let e):
- let v: UInt64 = try await read(UInt64.self, endian: e)
- if v > UInt64(cfg.maxFrameBytes) { throw NetSocketError.framingExceeded(max: cfg.maxFrameBytes) }
- length = Int(v)
- }
- if length > cfg.maxFrameBytes { throw NetSocketError.framingExceeded(max: cfg.maxFrameBytes) }
- return try await read(length)
- }
-
- /// Send an encodable value as a length-prefixed frame
+ /// Read data until a delimiter is found
///
- /// Uses the configured encoder (default: JSON) to serialize the value.
+ /// Searches the buffer for the delimiter pattern and returns all data up to (and optionally including)
+ /// the delimiter. The delimiter is always consumed from the stream.
///
/// - Parameters:
- /// - value: Value to encode and send
- /// - prefix: Length prefix type (default: u32 big-endian)
- /// - Throws: `NetSocketError.encodeFailed` if encoding fails
- public func send<T: Encodable>(_ value: T, prefix: LengthPrefix = .u32()) async throws {
- do {
- let data = try encodeValue(value)
- try await sendFrame(data, prefix: prefix)
- } catch {
- throw NetSocketError.encodeFailed(error)
+ /// - delimiter: Binary delimiter pattern to search for
+ /// - maxBytes: Maximum bytes to read before throwing (default: no limit)
+ /// - includeDelimiter: Whether to include delimiter in result (default: false)
+ /// - Returns: Data read from stream
+ /// - Throws: `NetSocketError.framingExceeded` if max bytes exceeded, or connection errors
+ public func read(past delimiter: Data, maxBytes: Int? = nil, includeDelimiter: Bool = false) async throws -> Data {
+ while true {
+ try Task.checkCancellation()
+ if let r = search(delimiter: delimiter) {
+ let consumeLen = r.upperBound - head
+ let data = try await read(consumeLen)
+ return includeDelimiter ? data : data.dropLast(delimiter.count)
+ }
+ if let maxBytes, availableBytes >= maxBytes {
+ throw NetSocketError.framingExceeded(max: maxBytes)
+ }
+ try await waitForData()
+ guard !isClosed || availableBytes > 0 else { throw NetSocketError.closed }
}
}
- /// Receive and decode a length-prefixed value
+ /// Read exactly N bytes from the socket
///
- /// Uses the configured decoder (default: JSON) to deserialize the value.
+ /// Waits for data to arrive if buffer doesn't contain enough bytes yet. The internal buffer
+ /// is automatically compacted after reading to prevent unbounded memory growth.
///
- /// - Parameters:
- /// - type: Type to decode
- /// - prefix: Length prefix type (default: u32 big-endian)
- /// - Returns: Decoded value
- /// - Throws: `NetSocketError.decodeFailed` if decoding fails
- public func receive<T: Decodable>(_ type: T.Type, prefix: LengthPrefix = .u32()) async throws -> T {
- let data = try await receiveFrame(prefix: prefix)
- do {
- let decoded = try decodeValue(data, T.self)
- guard let result = decoded as? T else {
- throw NetSocketError.decodeFailed(NSError(
- domain: "NetSocketNew",
- code: -1,
- userInfo: [NSLocalizedDescriptionKey: "Type mismatch in decode"]
- ))
- }
- return result
- } catch {
- throw NetSocketError.decodeFailed(error)
- }
+ /// - Parameter count: Number of bytes to read
+ /// - Returns: Exactly `count` bytes
+ /// - Throws: `NetSocketError.insufficientData` if connection closes before enough data arrives
+ public func read(_ count: Int) async throws -> Data {
+ try await self.ensureReadable(count)
+ let start = self.head
+ let end = self.head + count
+ let slice = self.buffer[start..<end]
+ self.head = end
+ self.compactIfNeeded()
+ return Data(slice)
}
-
- // MARK: Read typed & utilities
-
+
/// Read a fixed-width integer from the socket
///
/// - Parameters:
@@ -595,7 +346,7 @@ public actor NetSocketNew {
/// - Throws: `NetSocketError` if insufficient data or connection closed
public func read<T: FixedWidthInteger>(_ type: T.Type = T.self, endian: Endian = .big) async throws -> T {
let size = MemoryLayout<T>.size
- let data = try await read(size)
+ let data = try await self.read(size)
let value: T = data.withUnsafeBytes { raw in
raw.load(as: T.self)
}
@@ -613,8 +364,10 @@ public actor NetSocketNew {
/// - Returns: Decoded string
/// - Throws: `NetSocketError` if decoding fails or insufficient data
public func read(_ length: Int, encoding: String.Encoding = .utf8) async throws -> String {
- let data = try await read(length)
- guard let s = String(data: data, encoding: encoding) else { throw NetSocketError.decodeFailed(NSError()) }
+ let data = try await self.read(length)
+ guard let s = String(data: data, encoding: encoding) else {
+ throw NetSocketError.decodeFailed(NSError())
+ }
return s
}
@@ -632,118 +385,6 @@ public actor NetSocketNew {
return s
}
- /// Read a pascal string (1-byte length prefix followed by string data)
- ///
- /// This method reads a single byte for the length, then reads that many bytes and attempts
- /// to decode them as a string. It tries multiple encodings for compatibility with legacy
- /// protocols like Hotline: UTF-8, Shift-JIS, Windows-1251, and falls back to MacRoman.
- ///
- /// - Returns: The decoded string, or nil if length is 0
- /// - Throws: `NetSocketError` if reading fails or no encoding succeeds
- public func readPascalString() async throws -> String? {
- let length = try await read(UInt8.self)
- guard length > 0 else { return nil }
-
- let data = try await read(Int(length))
-
- // Try auto-detection with common encodings
- let allowedEncodings = [
- String.Encoding.utf8.rawValue,
- String.Encoding.shiftJIS.rawValue,
- String.Encoding.unicode.rawValue,
- String.Encoding.windowsCP1251.rawValue
- ]
-
- var decodedString: NSString?
- let detected = NSString.stringEncoding(
- for: data,
- encodingOptions: [.allowLossyKey: false],
- convertedString: &decodedString,
- usedLossyConversion: nil
- )
-
- if allowedEncodings.contains(detected), let str = decodedString as? String {
- return str
- }
-
- // Fallback to MacRoman for classic Mac compatibility
- guard let str = String(data: data, encoding: .macOSRoman) else {
- throw NetSocketError.decodeFailed(NSError(
- domain: "NetSocketNew",
- code: -1,
- userInfo: [NSLocalizedDescriptionKey: "Failed to decode pascal string with any known encoding"]
- ))
- }
- return str
- }
-
- /// Read data until a delimiter is found
- ///
- /// Searches the buffer for the delimiter pattern and returns all data up to (and optionally including)
- /// the delimiter. The delimiter is always consumed from the stream.
- ///
- /// - Parameters:
- /// - delimiter: Binary delimiter pattern to search for
- /// - maxBytes: Maximum bytes to read before throwing (default: no limit)
- /// - includeDelimiter: Whether to include delimiter in result (default: false)
- /// - Returns: Data read from stream
- /// - Throws: `NetSocketError.framingExceeded` if max bytes exceeded, or connection errors
- public func read(past delimiter: Data, maxBytes: Int? = nil, includeDelimiter: Bool = false) async throws -> Data {
- while true {
- try Task.checkCancellation()
- if let r = search(delimiter: delimiter) {
- let consumeLen = r.upperBound - head
- let data = try await read(consumeLen)
- return includeDelimiter ? data : data.dropLast(delimiter.count)
- }
- if let maxBytes, availableBytes >= maxBytes {
- throw NetSocketError.framingExceeded(max: maxBytes)
- }
- try await waitForData()
- guard !isClosed || availableBytes > 0 else { throw NetSocketError.closed }
- }
- }
-
- /// Read exactly N bytes from the socket
- ///
- /// Waits for data to arrive if buffer doesn't contain enough bytes yet. The internal buffer
- /// is automatically compacted after reading to prevent unbounded memory growth.
- ///
- /// - Parameter count: Number of bytes to read
- /// - Returns: Exactly `count` bytes
- /// - Throws: `NetSocketError.insufficientData` if connection closes before enough data arrives
- public func read(_ count: Int) async throws -> Data {
- try await ensureReadable(count)
- let start = head
- let end = head + count
- let slice = buffer[start..<end]
- head = end
- compactIfNeeded()
- return Data(slice)
- }
-
- /// Skip/discard exactly N bytes from the stream without allocating memory
- public func skip(_ count: Int) async throws {
- guard count > 0 else { return }
- try await ensureReadable(count)
- head += count
- compactIfNeeded()
- }
-
- /// Skip until delimiter is found (discards delimiter too)
- public func skip(past delimiter: Data) async throws {
- while true {
- try Task.checkCancellation()
- if let r = search(delimiter: delimiter) {
- head = r.upperBound // Skip to end of delimiter
- compactIfNeeded()
- return
- }
- try await waitForData()
- guard !isClosed else { throw NetSocketError.closed }
- }
- }
-
/// Read exactly N bytes with progress callbacks
///
/// Like `read(_:)`, but reads in chunks and reports progress after each chunk.
@@ -782,236 +423,261 @@ public actor NetSocketNew {
return data
}
-
- func peek(_ count: Int) async throws -> Data {
- try await ensureReadable(count)
- let slice = buffer[head..<(head + count)]
- return Data(slice) // Don't advance head
- }
-
- // MARK: Internals
- private var availableBytes: Int { buffer.count - head }
+ // MARK: Peek Data
- private func waitForData() async throws {
- try Task.checkCancellation()
- try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
- if isClosed { cont.resume(); return }
- dataWaiters.append(cont)
+ public var availableBytes: Int { self.buffer.count - self.head }
+
+ public func peek(_ count: Int) -> Data? {
+ guard self.availableBytes >= count else {
+ return nil
}
+
+ let slice = self.buffer[self.head..<(self.head + count)]
+ return Data(slice) // Don't advance head
}
- private func ensureReadable(_ count: Int) async throws {
- try await ensureReady()
- while availableBytes < count {
- try Task.checkCancellation()
- if isClosed { throw NetSocketError.insufficientData(expected: count, got: availableBytes) }
- try await waitForData()
+ public func peek(upto count: Int) -> Data {
+ let amount = min(self.availableBytes, count)
+ guard amount > 0 else {
+ return Data()
}
+
+ let slice = self.buffer[self.head..<(self.head + amount)]
+ return Data(slice)
}
- private func ensureReady() async throws {
- if isClosed { throw NetSocketError.closed }
- if !ready { try await waitUntilReady() }
+ public func peek(awaiting count: Int) async throws -> Data {
+ try await self.ensureReadable(count)
+ let slice = self.buffer[self.head..<(self.head + count)]
+ return Data(slice) // Don't advance head
}
- private func compactIfNeeded() {
- // Avoid unbounded memory as head advances
- if head > 64 * 1024 && head > buffer.count / 2 {
- buffer.removeSubrange(0..<head)
- head = 0
- }
+ // MARK: Skip Data
+
+ /// Skip/discard exactly N bytes from the stream without allocating memory
+ public func skip(_ count: Int) async throws {
+ guard count > 0 else { return }
+ try await self.ensureReadable(count)
+ self.head += count
+ self.compactIfNeeded()
}
- private func search(delimiter: Data) -> Range<Int>? {
- guard !delimiter.isEmpty, availableBytes >= delimiter.count else { return nil }
- let hay = buffer[head..<buffer.count]
-
- // Fast path for single-byte delimiters
- if delimiter.count == 1, let byte = delimiter.first {
- if let idx = hay.firstIndex(of: byte) {
- let pos = head + hay.distance(from: hay.startIndex, to: idx)
- return pos..<(pos + 1)
+ /// Skip until delimiter is found (discards delimiter too)
+ public func skip(past delimiter: Data) async throws {
+ while true {
+ try Task.checkCancellation()
+ if let r = self.search(delimiter: delimiter) {
+ self.head = r.upperBound // Skip to end of delimiter
+ self.compactIfNeeded()
+ return
+ }
+ try await self.waitForData()
+ guard !self.isClosed else {
+ throw NetSocketError.closed
}
- return nil
- }
-
- // General case
- if let r = hay.firstRange(of: delimiter) {
- let lower = head + hay.distance(from: hay.startIndex, to: r.lowerBound)
- let upper = head + hay.distance(from: hay.startIndex, to: r.upperBound)
- return lower..<upper
- }
-
- return nil
- }
-}
-
-// MARK: - Small helpers
-
-private extension Data {
- mutating func appendInteger<T: FixedWidthInteger>(_ value: T, endian: Endian) throws {
- var v = value
- switch endian {
- case .big: v = T(bigEndian: value)
- case .little: v = T(littleEndian: value)
- }
- var copy = v
- withUnsafePointer(to: &copy) { ptr in
- self.append(contentsOf: UnsafeRawBufferPointer(start: ptr, count: MemoryLayout<T>.size))
}
}
-}
-
-public extension NetSocketNew {
- /// Progress information for file uploads/downloads
- struct FileProgress: Sendable {
- /// Number of bytes sent/received so far
- public let sent: Int64
- /// Total file size (may be nil if unknown)
- public let total: Int64?
- }
-
- /// Upload a file to the socket without framing (raw byte stream)
+
+ // MARK: Files
+
+ /// Upload a file from a URL, yielding progress as an AsyncSequence.
///
- /// Reads and writes the file in chunks to limit memory usage. Each chunk waits for network
- /// backpressure via `.contentProcessed` before reading the next chunk.
+ /// Iterating this sequence drives the transfer. Each yielded value reports
+ /// the total bytes sent so far and the known total. Cancel the consuming
+ /// task to cancel the transfer.
+ ///
+ /// This method handles opening and closing the file handle automatically.
///
/// - Parameters:
- /// - url: File URL to upload
- /// - chunkSize: Chunk size for reading/writing (default: 256 KB)
- /// - progress: Optional progress callback
- /// - Returns: Total bytes sent
- /// - Throws: File I/O or network errors
- @discardableResult
- func writeFile(
- from url: URL,
- chunkSize: Int = 256 * 1024,
- progress: (@Sendable (FileProgress) -> Void)? = nil
- ) async throws -> Int64 {
- try await ensureReady()
- let total = try? self.fileLength(at: url)
-
- let fh = try FileHandle(forReadingFrom: url)
- defer { try? fh.close() }
-
- var sent: Int64 = 0
- while true {
- try Task.checkCancellation()
- guard let chunk = try fh.read(upToCount: chunkSize), !chunk.isEmpty else { break }
- try await write(chunk) // uses .contentProcessed completion inside
- sent += Int64(chunk.count)
- progress?(.init(sent: sent, total: total))
+ /// - url: File URL to upload.
+ /// - chunkSize: Size of each read chunk.
+ /// - Returns: An `AsyncThrowingStream` of `FileProgress` updates.
+ func writeFile(from url: URL, chunkSize: Int = 256 * 1024) -> AsyncThrowingStream<FileProgress, Error> {
+ // This stream wrapper manages the FileHandle's lifetime.
+ return AsyncThrowingStream(bufferingPolicy: .bufferingOldest(1)) { continuation in
+ // Capture self (the actor) to use in detached task
+ let actor = self
+
+ // Open file on a background thread (file I/O is blocking)
+ let task = Task.detached {
+ let fh: FileHandle
+ let total: Int
+
+ // 1. Open file and get length (blocking I/O, done off-actor)
+ do {
+ total = Int(try NetSocketNew.fileLength(at: url))
+ fh = try FileHandle(forReadingFrom: url)
+ } catch {
+ continuation.finish(throwing: NetSocketError.failed(underlying: error))
+ return
+ }
+
+ // 2. Now switch to the actor context to call the actor-isolated method
+ let stream = await actor.writeFile(
+ from: fh,
+ length: total,
+ chunkSize: chunkSize
+ )
+
+ // 3. Forward all elements from the underlying stream to our stream
+ do {
+ for try await progress in stream {
+ try Task.checkCancellation() // Exit early if cancelled
+ continuation.yield(progress)
+ }
+ try? fh.close()
+ continuation.finish()
+ } catch is CancellationError {
+ try? fh.close()
+ continuation.finish()
+ } catch {
+ try? fh.close()
+ continuation.finish(throwing: error)
+ }
+ }
+
+ // If the *consumer* cancels the stream, we cancel our managing task.
+ continuation.onTermination = { @Sendable _ in
+ task.cancel()
+ }
}
- return sent
}
- /// Upload a file as a length-prefixed frame without buffering the entire file in memory
+ /// Upload a file from an open FileHandle, yielding progress as an AsyncSequence.
///
- /// Sends the file size as a length prefix, then streams the file content in chunks.
- /// Memory-efficient for large files.
+ /// Iterating this sequence drives the transfer. Each yielded value reports
+ /// the total bytes sent so far and the known total. Cancel the consuming
+ /// task to cancel the transfer.
+ ///
+ /// **Note:** The caller is responsible for opening and closing the `fileHandle`.
///
/// - Parameters:
- /// - url: File URL to upload
- /// - lengthPrefix: Length prefix type (default: u64 big-endian)
- /// - chunkSize: Chunk size for reading/writing (default: 256 KB)
- /// - progress: Optional progress callback
- /// - Returns: Total bytes sent (not including length header)
- /// - Throws: File I/O, framing, or network errors
- @discardableResult
- func sendFileFramed(
- _ url: URL,
- lengthPrefix: LengthPrefix = .u64(.big),
- chunkSize: Int = 256 * 1024,
- progress: (@Sendable (FileProgress) -> Void)? = nil
- ) async throws -> Int64 {
- let total = try fileLength(at: url)
- try ensure(total, fitsIn: lengthPrefix)
+ /// - fileHandle: Open `FileHandle` for reading.
+ /// - length: Exact number of bytes to send (total file size).
+ /// - chunkSize: Size of each read chunk.
+ /// - Returns: An `AsyncThrowingStream` of `FileProgress` updates.
+ func writeFile(from fileHandle: FileHandle, length: Int, chunkSize: Int = 256 * 1024) -> AsyncThrowingStream<FileProgress, Error> {
+ precondition(length >= 0, "length must be >= 0")
- // 1) Send the length header
- var header = Data()
- switch lengthPrefix {
- case .u8:
- header.append(UInt8(truncatingIfNeeded: total))
- case .u16(let e):
- try header.appendInteger(UInt16(truncatingIfNeeded: total), endian: e)
- case .u32(let e):
- try header.appendInteger(UInt32(truncatingIfNeeded: total), endian: e)
- case .u64(let e):
- try header.appendInteger(UInt64(total), endian: e)
+ if length == 0 {
+ return AsyncThrowingStream { continuation in
+ continuation.yield(.init(sent: 0, total: 0, bytesPerSecond: 0, estimatedTimeRemaining: 0))
+ continuation.finish()
+ }
}
- try await write(header)
- // 2) Stream the file bytes (raw) right after the header
- let sent = try await writeFile(from: url, chunkSize: chunkSize) { prog in
- progress?(prog)
+ return AsyncThrowingStream(bufferingPolicy: .bufferingOldest(1)) { continuation in
+ let task = Task { [weak self] in
+ guard let self else {
+ continuation.finish()
+ return
+ }
+
+ var estimator = TransferRateEstimator(total: Int(length))
+
+ do {
+ try await self.ensureReady()
+
+ while estimator.transferred < length {
+ try Task.checkCancellation()
+
+ let toRead = Int(min(chunkSize, length - estimator.transferred))
+
+ // Read from disk
+ guard let chunk = try fileHandle.read(upToCount: toRead), !chunk.isEmpty else {
+ if estimator.transferred < length {
+ throw NetSocketError.failed(underlying: NSError(
+ domain: "NetSocket", code: 9001,
+ userInfo: [NSLocalizedDescriptionKey: "File read ended prematurely. Expected \(length) bytes, got \(estimator.transferred)."]
+ ))
+ }
+ break
+ }
+
+ // Write to network
+ try await self.write(chunk)
+
+ // Update estimator and yield progress
+ let progress = estimator.update(bytes: chunk.count)
+ continuation.yield(progress)
+ }
+
+ continuation.finish()
+ } catch {
+ continuation.finish(throwing: error)
+ }
+ }
+
+ continuation.onTermination = { @Sendable _ in
+ task.cancel()
+ }
}
- return sent
}
- /// Download a length-prefixed file and write it to disk in chunks (bounded memory)
+ /// Receive a file of known length and yield progress updates as an AsyncSequence.
///
- /// Reads the file size from a length prefix, then streams the content directly to disk
- /// in chunks to avoid loading the entire file into memory.
+ /// Iterating this sequence drives the transfer. Each yielded value reports
+ /// the total bytes written so far and the known total. Cancel the consuming
+ /// task to cancel the transfer.
///
/// - Parameters:
- /// - url: Destination file URL
- /// - lengthPrefix: Length prefix type (default: u64 big-endian)
- /// - chunkSize: Chunk size for reading/writing (default: 256 KB)
- /// - overwrite: Whether to overwrite existing file (default: true)
- /// - progress: Optional progress callback
- /// - Returns: Total bytes written
- /// - Throws: File I/O, framing, or network errors
- @discardableResult
- func receiveFile(
- to url: URL,
- lengthPrefix: LengthPrefix = .u64(.big),
- chunkSize: Int = 256 * 1024,
- overwrite: Bool = true,
- progress: (@Sendable (FileProgress) -> Void)? = nil
- ) async throws -> Int64 {
- // 1) Read length header
- let total64: Int64 = try await {
- switch lengthPrefix {
- case .u8: return Int64(try await read(UInt8.self))
- case .u16(let e): return Int64(try await read(UInt16.self, endian: e))
- case .u32(let e): return Int64(try await read(UInt32.self, endian: e))
- case .u64(let e):
- let v: UInt64 = try await read(UInt64.self, endian: e)
- guard v <= UInt64(Int64.max) else {
- throw NetSocketError.framingExceeded(max: Int(Int64.max))
- }
- return Int64(v)
- }
- }()
-
- // 2) Prepare destination file
- if overwrite { try? FileManager.default.removeItem(at: url) }
- FileManager.default.createFile(atPath: url.path, contents: nil, attributes: nil)
- let fh = try FileHandle(forWritingTo: url)
- defer { try? fh.close() }
+ /// - fileHandle: Open `FileHandle` for writing (caller must close).
+ /// - length: Exact number of bytes expected.
+ /// - chunkSize: Size of each read chunk.
+ /// - Returns: An `AsyncThrowingStream` of `FileProgress` updates.
+ func receiveFile(to fileHandle: FileHandle, length: Int, chunkSize: Int = 256 * 1024) -> AsyncThrowingStream<FileProgress, Error> {
+ precondition(length >= 0, "length must be >= 0")
- // 3) Stream chunks from the socket into the file
- var remaining = total64
- var written: Int64 = 0
+ if length == 0 {
+ return AsyncThrowingStream { continuation in
+ continuation.yield(.init(sent: 0, total: 0, bytesPerSecond: 0, estimatedTimeRemaining: 0))
+ continuation.finish()
+ }
+ }
- while remaining > 0 {
- try Task.checkCancellation()
- let n = Int(min(Int64(chunkSize), remaining))
- let chunk = try await read(n) // reuses your internal buffer, bounded by n
- fh.write(chunk)
- remaining -= Int64(n)
- written += Int64(n)
- progress?(.init(sent: written, total: total64))
+ return AsyncThrowingStream(bufferingPolicy: .bufferingOldest(1)) { continuation in
+ let task = Task { [weak self] in
+ guard let self else {
+ continuation.finish()
+ return
+ }
+
+ var estimator = TransferRateEstimator(total: length)
+
+ do {
+ var remaining: Int = length
+
+ while remaining > 0 {
+ try Task.checkCancellation()
+ let n = min(chunkSize, remaining)
+
+ let chunk = try await self.read(n)
+ try fileHandle.write(contentsOf: chunk)
+
+ let chunkSize = Int(chunk.count)
+ remaining -= chunkSize
+ let progress = estimator.update(bytes: chunkSize)
+ continuation.yield(progress)
+ }
+
+ continuation.finish()
+ } catch {
+ continuation.finish(throwing: error)
+ }
+ }
+
+ continuation.onTermination = { @Sendable _ in
+ task.cancel()
+ }
}
- return written
}
-
+
/// Download a file of known length and write it to disk in chunks
///
- /// Unlike `receiveFile()`, this method does **not** read a length prefix. The caller must
- /// provide the expected file size (e.g., from protocol metadata). The file is streamed
- /// directly to disk to avoid loading it entirely into memory.
+ /// This method does **not** read a length prefix. The caller must provide the expected
+ /// file size (e.g., from protocol metadata). The file is streamed directly to disk to
+ /// avoid loading it entirely into memory.
///
/// Supports atomic writes: when enabled, data is written to a temporary `.part` file and
/// renamed on success. If an error occurs, the temporary file is automatically cleaned up.
@@ -1030,60 +696,53 @@ public extension NetSocketNew {
/// ```swift
/// // Hotline protocol: file size comes from transaction header
/// let transaction = try await socket.receive(HotlineTransaction.self)
- /// try await socket.receiveFileKnownLength(
+ /// try await socket.receiveFile(
/// to: destinationURL,
/// length: transaction.fileSize
/// )
/// ```
@discardableResult
- func receiveFileKnownLength(
+ func receiveFile(
to url: URL,
- length: Int64,
+ length: Int,
chunkSize: Int = 256 * 1024,
overwrite: Bool = true,
atomic: Bool = true,
progress: (@Sendable (FileProgress) -> Void)? = nil
- ) async throws -> Int64 {
+ ) async throws -> Int {
precondition(length >= 0, "length must be >= 0")
-
- // Validate length doesn't exceed configured maximum
- guard length <= cfg.maxFrameBytes else {
- throw NetSocketError.framingExceeded(max: cfg.maxFrameBytes)
- }
-
+
// Fast path: nothing to do
if length == 0 {
if overwrite { try? FileManager.default.removeItem(at: url) }
FileManager.default.createFile(atPath: url.path, contents: Data(), attributes: nil)
return 0
}
-
+
// Prepare destination (optionally atomic)
let fm = FileManager.default
let dir = url.deletingLastPathComponent()
- let tmp = atomic
- ? dir.appendingPathComponent(".\(url.lastPathComponent).part-\(UUID().uuidString)")
- : url
-
+ let tmp = atomic ? dir.appendingPathComponent(".\(url.lastPathComponent).part-\(UUID().uuidString)") : url
+
if overwrite { try? fm.removeItem(at: tmp) }
if overwrite, !atomic { try? fm.removeItem(at: url) }
-
+
// Create and open the file for writing
fm.createFile(atPath: tmp.path, contents: nil, attributes: nil)
let fh = try FileHandle(forWritingTo: tmp)
defer { try? fh.close() }
-
- var remaining = length
- var written: Int64 = 0
-
+
+ var remaining: Int = length
+ var written: Int = 0
+
do {
while remaining > 0 {
try Task.checkCancellation()
- let n = Int(min(Int64(chunkSize), remaining))
- let chunk = try await read(n)
- fh.write(chunk)
- remaining -= Int64(n)
- written += Int64(n)
+ let n = Int(min(chunkSize, remaining))
+ let chunk = try await self.read(n)
+ try fh.write(contentsOf: chunk)
+ remaining -= n
+ written += Int(n)
progress?(.init(sent: written, total: length))
}
} catch {
@@ -1091,20 +750,162 @@ public extension NetSocketNew {
if atomic { try? fm.removeItem(at: tmp) }
throw error
}
-
+
// Atomically move into place if requested
if atomic {
if overwrite { try? fm.removeItem(at: url) }
try fm.moveItem(at: tmp, to: url)
}
-
+
return written
}
-}
-
-// MARK: - Small helpers (private)
-fileprivate extension NetSocketNew {
- func fileLength(at url: URL) throws -> Int64 {
+
+ // MARK: Internals
+
+ private func start() async throws {
+ self.connection.stateUpdateHandler = { state in
+ Task { [weak self] in
+ guard let self else { return }
+ switch state {
+ case .ready:
+ await self.setReady()
+ await self.resumeReadyWaiters(with: .success(()))
+ case .failed(let error):
+ await self.failAllWaiters(NetSocketError.failed(underlying: error))
+ await self.setClosed()
+ case .waiting(let error):
+ // bubble as transient failure for awaiters; reconnect logic could live here
+ await self.resumeReadyWaiters(with: .failure(NetSocketError.failed(underlying: error)))
+ case .cancelled:
+ await self.failAllWaiters(NetSocketError.closed)
+ await self.setClosed()
+ default:
+ break
+ }
+ }
+ }
+
+ // Kick off receive loop after .start
+ self.connection.start(queue: queue)
+ try await self.waitUntilReady()
+ self.startReceiveLoop()
+ }
+
+ private func startReceiveLoop() {
+ @Sendable func loop(_ connection: NWConnection, chunk: Int, owner: NetSocketNew, connID: String) {
+ print("NetSocketNew[\(connID)]: Calling connection.receive(\(chunk)) to request more data...")
+
+ connection.receive(minimumIncompleteLength: 1, maximumLength: chunk) { [weak owner] data, _, isComplete, error in
+ print("NetSocketNew[\(connID)]: Receive callback - data: \(data?.count ?? 0) bytes, isComplete: \(isComplete), error: \(String(describing: error))")
+ Task {
+ guard let o = owner else {
+ return
+ }
+
+ if let error {
+ await o.handleReceiveError(error)
+ return
+ }
+ if let data, !data.isEmpty {
+ await o.append(data, connID: connID)
+ }
+ if isComplete {
+ print("NetSocketNew[\(connID)]: EOF from peer.")
+ await o.handleEOF()
+ return
+ }
+ loop(connection, chunk: chunk, owner: o, connID: connID)
+ }
+ }
+ }
+ loop(connection, chunk: self.config.receiveChunk, owner: self, connID: connectionID)
+ }
+
+ private func handleReceiveError(_ error: Error) {
+ self.isClosed = true
+ self.failAllWaiters(NetSocketError.failed(underlying: error))
+ }
+
+ private func handleEOF() {
+ self.isClosed = true
+ let waiters = self.dataWaiters
+ self.dataWaiters.removeAll()
+ for w in waiters {
+ w.resume()
+ } // wake so readers can observe closure
+ }
+
+ private func setReady() {
+ self.ready = true
+ }
+
+ private func setClosed() {
+ self.isClosed = true
+ }
+
+ private func ensureReady() async throws {
+ if self.isClosed {
+ throw NetSocketError.closed
+ }
+ if !self.ready {
+ try await self.waitUntilReady()
+ }
+ }
+
+ private func ensureReadable(_ count: Int) async throws {
+ try await self.ensureReady()
+ while self.availableBytes < count {
+ try Task.checkCancellation()
+ if self.isClosed {
+ throw NetSocketError.insufficientData(expected: count, got: self.availableBytes)
+ }
+ try await self.waitForData()
+ }
+ }
+
+ private func waitForData() async throws {
+ try Task.checkCancellation()
+ try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
+ if self.isClosed {
+ cont.resume()
+ return
+ }
+ self.dataWaiters.append(cont)
+ }
+ }
+
+ private func compactIfNeeded() {
+ // Avoid unbounded memory as head advances
+ if self.head > 64 * 1024 && self.head > self.buffer.count / 2 {
+ self.buffer.removeSubrange(0..<self.head)
+ self.head = 0
+ }
+ }
+
+ private func search(delimiter: Data) -> Range<Int>? {
+ guard !delimiter.isEmpty, availableBytes >= delimiter.count else { return nil }
+ let hay = buffer[head..<buffer.count]
+
+ // Fast path for single-byte delimiters
+ if delimiter.count == 1, let byte = delimiter.first {
+ if let idx = hay.firstIndex(of: byte) {
+ let pos = head + hay.distance(from: hay.startIndex, to: idx)
+ return pos..<(pos + 1)
+ }
+ return nil
+ }
+
+ // General case
+ if let r = hay.firstRange(of: delimiter) {
+ let lower = head + hay.distance(from: hay.startIndex, to: r.lowerBound)
+ let upper = head + hay.distance(from: hay.startIndex, to: r.upperBound)
+ return lower..<upper
+ }
+
+ return nil
+ }
+
+ private static func fileLength(at url: URL) throws -> Int64 {
let values = try url.resourceValues(forKeys: [.isRegularFileKey, .fileSizeKey])
guard values.isRegularFile == true else {
throw NetSocketError.failed(underlying: NSError(
@@ -1114,29 +915,79 @@ fileprivate extension NetSocketNew {
}
if let s = values.fileSize { return Int64(s) }
let attrs = try FileManager.default.attributesOfItem(atPath: url.path)
- if let n = attrs[.size] as? NSNumber { return n.int64Value }
+ if let n = attrs[.size] as? NSNumber {
+ return n.int64Value
+ }
throw NetSocketError.failed(underlying: NSError(
domain: "NetSocket", code: 1002,
userInfo: [NSLocalizedDescriptionKey: "Unable to determine file size for \(url.lastPathComponent)"]
))
}
- func ensure(_ length: Int64, fitsIn prefix: LengthPrefix) throws {
- let max: Int64 = {
- switch prefix {
- case .u8: return Int64(UInt8.max)
- case .u16: return Int64(UInt16.max)
- case .u32: return Int64(UInt32.max)
- case .u64: return Int64.max
+ private func waitUntilReady() async throws {
+ guard !self.ready else { return }
+ try await withCheckedThrowingContinuation { (cont: CheckedContinuation<Void, Error>) in
+ self.readyWaiters.append(cont)
+ }
+ }
+
+ private func resumeReadyWaiters(with result: Result<Void, Error>) {
+ let waiters = self.readyWaiters
+ self.readyWaiters.removeAll()
+ for w in waiters {
+ switch result {
+ case .success: w.resume()
+ case .failure(let e): w.resume(throwing: e)
}
- }()
- if length > max {
- throw NetSocketError.framingExceeded(max: Int(max))
+ }
+ }
+
+ private func failAllWaiters(_ error: Error) {
+ self.resumeReadyWaiters(with: .failure(error))
+ let waiters = self.dataWaiters
+ self.dataWaiters.removeAll()
+ for w in waiters {
+ w.resume(throwing: error)
+ }
+ }
+
+ private func append(_ data: Data, connID: String) {
+ print("NetSocketNew[\(connID)]: Received \(data.count) bytes from network, buffer now has \(buffer.count - head + data.count) available")
+ buffer.append(data)
+ if buffer.count - head > config.maxBufferBytes {
+ // Hard stop: drop connection rather than OOM'ing.
+ isClosed = true
+ connection.cancel()
+ failAllWaiters(NetSocketError.framingExceeded(max: config.maxBufferBytes))
+ return
+ }
+ resumeDataWaiters()
+ }
+
+ private func resumeDataWaiters() {
+ let waiters = dataWaiters
+ dataWaiters.removeAll()
+ for w in waiters { w.resume() }
+ }
+}
+
+// MARK: - Utilities
+
+private extension Data {
+ mutating func appendInteger<T: FixedWidthInteger>(_ value: T, endian: Endian) throws {
+ var v = value
+ switch endian {
+ case .big: v = T(bigEndian: value)
+ case .little: v = T(littleEndian: value)
+ }
+ var copy = v
+ withUnsafePointer(to: &copy) { ptr in
+ self.append(contentsOf: UnsafeRawBufferPointer(start: ptr, count: MemoryLayout<T>.size))
}
}
}
-// MARK: - Stream-based Encoding/Decoding
+// MARK: - NetSocketEncodable
/// Protocol for types that can encode themselves to binary data
///
@@ -1243,7 +1094,7 @@ public extension NetSocketNew {
/// - Throws: Encoding or network errors
func send<T: NetSocketEncodable>(_ value: T, endian: Endian = .big) async throws {
let data = try value.encode(endian: endian)
- try await write(data)
+ try await self.write(data)
}
/// Receive and decode a value directly from the socket stream (no length prefix)
diff --git a/Hotline/Library/NetSocket/TransferRateEstimator.swift b/Hotline/Library/NetSocket/TransferRateEstimator.swift
new file mode 100644
index 0000000..7d16904
--- /dev/null
+++ b/Hotline/Library/NetSocket/TransferRateEstimator.swift
@@ -0,0 +1,135 @@
+// TransferRateEstimator
+// Dustin Mierau • @mierau
+// MIT License
+
+import Foundation
+
+/// Transfer rate estimator using exponential moving average (EMA)
+///
+/// Tracks transfer speed and estimates time remaining. Designed to smooth out
+/// network jitter and provide stable estimates after collecting enough samples.
+///
+/// Example:
+/// ```swift
+/// var estimator = TransferRateEstimator(total: fileSize)
+///
+/// while transferring {
+/// let chunk = try await receiveData()
+/// let progress = estimator.update(bytes: chunk.count)
+/// print("Speed: \(progress.bytesPerSecond ?? 0) B/s, ETA: \(progress.estimatedTimeRemaining ?? 0)s")
+/// }
+/// ```
+public struct TransferRateEstimator {
+ /// Total bytes to transfer (nil if unknown)
+ public let total: Int?
+
+ /// Exponential moving average of transfer rate (bytes/second)
+ private var emaBytesPerSecond: Double = 0
+
+ /// Smoothing factor for EMA (0 < alpha ≤ 1)
+ /// Higher = more responsive to recent changes, lower = more smoothing
+ private let alpha: Double
+
+ /// Number of samples collected
+ private var sampleCount: Int = 0
+
+ /// Timestamp of first sample (for elapsed time calculation)
+ private var startTime: ContinuousClock.Instant?
+
+ /// Timestamp of last update (for calculating sample duration)
+ private var lastUpdateTime: ContinuousClock.Instant?
+
+ /// Minimum elapsed time before trusting estimates (seconds)
+ private let minElapsedTime: TimeInterval
+
+ /// Minimum number of samples before trusting estimates
+ private let minSamples: Int
+
+ /// Current number of bytes transferred
+ public private(set) var transferred: Int = 0
+
+ /// Create a new transfer rate estimator
+ ///
+ /// - Parameters:
+ /// - total: Total bytes to transfer (nil if unknown)
+ /// - alpha: EMA smoothing factor (default: 0.2). Range: 0.0-1.0
+ /// - minElapsedTime: Minimum elapsed time before estimates are reliable (default: 2.0s)
+ /// - minSamples: Minimum samples before estimates are reliable (default: 4)
+ public init(
+ total: Int? = nil,
+ alpha: Double = 0.2,
+ minElapsedTime: TimeInterval = 2.0,
+ minSamples: Int = 8
+ ) {
+ precondition(alpha > 0 && alpha <= 1, "alpha must be in range (0, 1]")
+ precondition(minSamples >= 0, "minSamples must be >= 0")
+
+ self.total = total
+ self.alpha = alpha
+ self.minElapsedTime = minElapsedTime
+ self.minSamples = minSamples
+ }
+
+ public mutating func update(total: Int) -> NetSocketNew.FileProgress {
+ return self.update(bytes: max(0, total - self.transferred))
+ }
+
+ /// Update the estimator with a new data sample
+ ///
+ /// Automatically calculates the duration since the last update.
+ ///
+ /// - Parameter bytes: Number of bytes transferred in this sample
+ /// - Returns: Current progress with speed and ETA estimates
+ public mutating func update(bytes: Int) -> NetSocketNew.FileProgress {
+ let clock = ContinuousClock()
+ let now = clock.now
+
+ // Record start time on first sample
+ if self.startTime == nil {
+ self.startTime = now
+ }
+
+ // Calculate duration since last update
+ let duration = self.lastUpdateTime.map { now - $0 } ?? .zero
+ self.lastUpdateTime = now
+
+ // Update transferred count
+ self.transferred += bytes
+
+ // Calculate instantaneous rate for this sample
+ let seconds: Double = duration / .seconds(1.0)
+ if seconds > 0 {
+ let instantRate = Double(bytes) / seconds
+ self.sampleCount += 1
+
+ // Update EMA
+ if self.emaBytesPerSecond == 0 {
+ self.emaBytesPerSecond = instantRate
+ } else {
+ self.emaBytesPerSecond += self.alpha * (instantRate - self.emaBytesPerSecond)
+ }
+ }
+
+ // Determine if we have enough data to trust the estimate
+ let elapsed = self.startTime.map { now - $0 } ?? .zero
+ let elapsedSeconds: Double = elapsed / .seconds(1.0)
+ let haveEstimate = (elapsedSeconds >= self.minElapsedTime || self.sampleCount >= self.minSamples) && self.emaBytesPerSecond > 0
+
+ // Calculate ETA if we have both an estimate and a known total
+ let eta: TimeInterval?
+ if haveEstimate, let total = self.total {
+ let remaining = total - self.transferred
+ eta = remaining > 0 ? TimeInterval(Double(remaining) / self.emaBytesPerSecond) : 0
+ } else {
+ eta = nil
+ }
+
+ return NetSocketNew.FileProgress(
+ sent: self.transferred,
+ total: self.total,
+ bytesPerSecond: haveEstimate ? self.emaBytesPerSecond : nil,
+ estimatedTimeRemaining: eta
+ )
+ }
+}
+
diff --git a/Hotline/Library/QuickLookPreviewView.swift b/Hotline/Library/QuickLookPreviewView.swift
new file mode 100644
index 0000000..6ba154e
--- /dev/null
+++ b/Hotline/Library/QuickLookPreviewView.swift
@@ -0,0 +1,22 @@
+import SwiftUI
+import Quartz
+
+/// Embeddable QuickLook preview view for macOS
+///
+/// This view uses QLPreviewView to display file previews inline, without showing a modal.
+/// Supports all file types that QuickLook supports (images, PDFs, videos, documents, etc.)
+struct QuickLookPreviewView: NSViewRepresentable {
+ let fileURL: URL
+
+ func makeNSView(context: Context) -> QLPreviewView {
+ let preview = QLPreviewView(frame: .zero, style: .normal)!
+ preview.autostarts = true
+ preview.shouldCloseWithWindow = true
+ preview.previewItem = fileURL as QLPreviewItem
+ return preview
+ }
+
+ func updateNSView(_ nsView: QLPreviewView, context: Context) {
+ nsView.previewItem = fileURL as QLPreviewItem
+ }
+}
diff --git a/Hotline/Library/URLAdditions.swift b/Hotline/Library/URLAdditions.swift
index 1f25541..0ba2100 100644
--- a/Hotline/Library/URLAdditions.swift
+++ b/Hotline/Library/URLAdditions.swift
@@ -1,4 +1,5 @@
import Foundation
+import UniformTypeIdentifiers
extension URL {
func generateUniqueFilePath(filename base: String) -> String {
@@ -24,3 +25,31 @@ extension URL {
return filePath
}
}
+
+extension UTType {
+ var canBePreviewedByQuickLook: Bool {
+ // QuickLook supports most common document types
+ let supportedSupertypes: [UTType] = [
+ .image,
+ .movie,
+ .audio,
+ .pdf,
+ .font,
+ .usdz,
+ .text,
+ .sourceCode,
+ .spreadsheet,
+ .presentation,
+
+// Microsoft Office
+ .init(filenameExtension: "doc")!,
+ .init(filenameExtension: "docx")!,
+ .init(filenameExtension: "xls")!,
+ .init(filenameExtension: "xlsx")!,
+ .init(filenameExtension: "ppt")!,
+ .init(filenameExtension: "pptx")!,
+ ]
+
+ return supportedSupertypes.contains { self.conforms(to: $0) }
+ }
+}
generated by cgit (git 2.54.0) at 2026-07-22 05:43:51 +0000