仓颉语言中Future/Promise模式：从异步回调到生产级并发框架

// 同步HTTP请求示例
func synchronousHttpRequests() {
    let urls = [
        "http://api1.example.com/data",
        "http://api2.example.com/data",
        "http://api3.example.com/data"
    ]
    
    let startTime = System.currentTimeMillis()
    let results = ArrayList<String>()
    
    // 串行执行，每个请求阻塞等待
    for (url in urls) {
        let response = httpGet(url)  // 阻塞3秒
        results.append(response)
    }
    
    let elapsed = System.currentTimeMillis() - startTime
    println("Total time: ${elapsed}ms")  // 约9000ms
}

// Promise状态枚举
public enum PromiseState {
    | Pending      // 待完成
    | Fulfilled    // 已完成（成功）
    | Rejected     // 已拒绝（失败）
}

// Promise结果类型
public enum PromiseResult<T> {
    | Success(T)
    | Failure(Exception)
}

// 核心Promise类
public class Promise<T> {
    private var state: PromiseState = PromiseState.Pending
    private var result: ?PromiseResult<T> = None
    private var callbacks: Array<(PromiseResult<T>) -> Unit>
    private let lock: Mutex
    
    public init() {
        this.callbacks = ArrayList<(PromiseResult<T>) -> Unit>()
        this.lock = Mutex()
    }
    
    // 完成Promise（成功）
    public func resolve(value: T) {
        lock.lock()
        
        if (state != PromiseState.Pending) {
            lock.unlock()
            return  // 已经完成，忽略
        }
        
        state = PromiseState.Fulfilled
        result = PromiseResult.Success(value)
        let callbacksCopy = callbacks.clone()
        
        lock.unlock()
        
        // 执行所有回调
        for (callback in callbacksCopy) {
            callback(result!)
        }
    }
    
    // 拒绝Promise（失败）
    public func reject(error: Exception) {
        lock.lock()
        
        if (state != PromiseState.Pending) {
            lock.unlock()
            return
        }
        
        state = PromiseState.Rejected
        result = PromiseResult.Failure(error)
        let callbacksCopy = callbacks.clone()
        
        lock.unlock()
        
        // 执行所有回调
        for (callback in callbacksCopy) {
            callback(result!)
        }
    }
    
    // 注册回调
    public func then(callback: (PromiseResult<T>) -> Unit) {
        lock.lock()
        
        if (state == PromiseState.Pending) {
            // 还未完成，注册回调
            callbacks.append(callback)
            lock.unlock()
        } else {
            // 已经完成，立即执行回调
            let resultCopy = result!
            lock.unlock()
            callback(resultCopy)
        }
    }
    
    // 获取当前状态
    public func getState(): PromiseState {
        lock.lock()
        defer { lock.unlock() }
        return state
    }
}

// Future类：Promise的只读视图
public class Future<T> {
    private let promise: Promise<T>
    
    // 从Promise创建Future
    internal init(promise: Promise<T>) {
        this.promise = promise
    }
    
    // 注册成功回调
    public func onSuccess(handler: (T) -> Unit): Future<T> {
        promise.then({ result =>
            match (result) {
                case Success(value) => handler(value)
                case Failure(_) => {}  // 忽略失败
            }
        })
        return this
    }
    
    // 注册失败回调
    public func onFailure(handler: (Exception) -> Unit): Future<T> {
        promise.then({ result =>
            match (result) {
                case Success(_) => {}  // 忽略成功
                case Failure(error) => handler(error)
            }
        })
        return this
    }
    
    // 注册完成回调（无论成功失败）
    public func onComplete(handler: (PromiseResult<T>) -> Unit): Future<T> {
        promise.then(handler)
        return this
    }
    
    // 链式转换
    public func map<U>(transform: (T) -> U): Future<U> {
        let newPromise = Promise<U>()
        
        this.onSuccess({ value =>
            try {
                let transformed = transform(value)
                newPromise.resolve(transformed)
            } catch (e: Exception) {
                newPromise.reject(e)
            }
        })
        
        this.onFailure({ error =>
            newPromise.reject(error)
        })
        
        return Future<U>(newPromise)
    }
    
    // 链式异步操作
    public func flatMap<U>(transform: (T) -> Future<U>): Future<U> {
        let newPromise = Promise<U>()
        
        this.onSuccess({ value =>
            try {
                let nextFuture = transform(value)
                nextFuture.onSuccess({ result =>
                    newPromise.resolve(result)
                })
                nextFuture.onFailure({ error =>
                    newPromise.reject(error)
                })
            } catch (e: Exception) {
                newPromise.reject(e)
            }
        })
        
        this.onFailure({ error =>
            newPromise.reject(error)
        })
        
        return Future<U>(newPromise)
    }
    
    // 阻塞等待结果（开发调试用）
    public func await(timeoutMs: Int64 = 30000): ?T {
        let startTime = System.currentTimeMillis()
        
        while (promise.getState() == PromiseState.Pending) {
            if (System.currentTimeMillis() - startTime > timeoutMs) {
                return None  // 超时
            }
            Thread.sleep(10)  // 短暂休眠
        }
        
        let result = promise.result!
        match (result) {
            case Success(value) => return value
            case Failure(error) => throw error
        }
    }
}

// 工作线程
private class WorkerThread {
    private let queue: BlockingQueue<() -> Unit>
    private var isRunning: Bool = true
    private let thread: Thread
    
    public init(queue: BlockingQueue<() -> Unit>) {
        this.queue = queue
        this.thread = Thread({ this.run() })
        this.thread.start()
    }
    
    private func run() {
        while (isRunning) {
            try {
                let task = queue.take()  // 阻塞等待任务
                task()
            } catch (e: Exception) {
                println("Worker thread error: ${e.getMessage()}")
            }
        }
    }
    
    public func stop() {
        isRunning = false
    }
}

// 线程池执行器
public class ThreadPoolExecutor {
    private let workers: Array<WorkerThread>
    private let taskQueue: BlockingQueue<() -> Unit>
    private let maxQueueSize: Int64
    private var isShutdown: Bool = false
    
    public init(poolSize: Int64 = 10, maxQueueSize: Int64 = 1000) {
        this.maxQueueSize = maxQueueSize
        this.taskQueue = BlockingQueue<() -> Unit>(maxQueueSize)
        this.workers = ArrayList<WorkerThread>()
        
        // 创建工作线程
        for (i in 0..poolSize) {
            workers.append(WorkerThread(taskQueue))
        }
    }
    
    // 提交任务，返回Future
    public func submit<T>(task: () -> T): Future<T> {
        if (isShutdown) {
            throw Exception("Executor is shutdown")
        }
        
        let promise = Promise<T>()
        
        let wrappedTask = {
            try {
                let result = task()
                promise.resolve(result)
            } catch (e: Exception) {
                promise.reject(e)
            }
        }
        
        taskQueue.offer(wrappedTask)
        
        return Future<T>(promise)
    }
    
    // 提交可能抛出异常的任务
    public func submitCallable<T>(callable: () throws -> T): Future<T> {
        if (isShutdown) {
            throw Exception("Executor is shutdown")
        }
        
        let promise = Promise<T>()
        
        let wrappedTask = {
            try {
                let result = callable()
                promise.resolve(result)
            } catch (e: Exception) {
                promise.reject(e)
            }
        }
        
        taskQueue.offer(wrappedTask)
        
        return Future<T>(promise)
    }
    
    // 关闭线程池
    public func shutdown() {
        isShutdown = true
        for (worker in workers) {
            worker.stop()
        }
    }
    
    // 获取队列大小
    public func getQueueSize(): Int64 {
        return taskQueue.size()
    }
}

// 延迟任务包装
private class DelayedTask {
    public let task: () -> Unit
    public let executeTime: Int64  // 执行时间戳
    
    public init(task: () -> Unit, delayMs: Int64) {
        this.task = task
        this.executeTime = System.currentTimeMillis() + delayMs
    }
    
    public func shouldExecute(): Bool {
        return System.currentTimeMillis() >= executeTime
    }
}

// 任务调度器
public class TaskScheduler {
    private let executor: ThreadPoolExecutor
    private let delayedTasks: PriorityQueue<DelayedTask>
    private let schedulerThread: Thread
    private var isRunning: Bool = true
    
    public init(poolSize: Int64 = 5) {
        this.executor = ThreadPoolExecutor(poolSize)
        this.delayedTasks = PriorityQueue<DelayedTask>({ a, b =>
            a.executeTime < b.executeTime
        })
        this.schedulerThread = Thread({ this.run() })
        this.schedulerThread.start()
    }
    
    // 立即执行任务
    public func execute<T>(task: () -> T): Future<T> {
        return executor.submit(task)
    }
    
    // 延迟执行任务
    public func schedule<T>(task: () -> T, delayMs: Int64): Future<T> {
        let promise = Promise<T>()
        
        let delayedTask = DelayedTask({
            try {
                let result = task()
                promise.resolve(result)
            } catch (e: Exception) {
                promise.reject(e)
            }
        }, delayMs)
        
        delayedTasks.offer(delayedTask)
        
        return Future<T>(promise)
    }
    
    // 定时执行任务
    public func scheduleAtFixedRate(task: () -> Unit, 
                                     initialDelayMs: Int64,
                                     periodMs: Int64) {
        let recursiveTask = {
            task()
            this.schedule({ task() }, periodMs)
        }
        
        this.schedule(recursiveTask, initialDelayMs)
    }
    
    // 调度器主循环
    private func run() {
        while (isRunning) {
            if (!delayedTasks.isEmpty()) {
                let nextTask = delayedTasks.peek()
                
                if (nextTask.shouldExecute()) {
                    delayedTasks.poll()
                    executor.submit({ nextTask.task() })
                }
            }
            
            Thread.sleep(10)  // 短暂休眠，避免CPU空转
        }
    }
    
    public func shutdown() {
        isRunning = false
        executor.shutdown()
    }
}

// Future工具类
public class Futures {
    // 等待所有Future完成
    public static func all<T>(futures: Array<Future<T>>): Future<Array<T>> {
        let promise = Promise<Array<T>>()
        let results = ArrayList<T>()
        var completedCount: Int64 = 0
        let lock = Mutex()
        
        for (future in futures) {
            future.onSuccess({ value =>
                lock.lock()
                results.append(value)
                completedCount += 1
                
                if (completedCount == Int64(futures.size)) {
                    promise.resolve(results.toArray())
                }
                lock.unlock()
            })
            
            future.onFailure({ error =>
                lock.lock()
                if (promise.getState() == PromiseState.Pending) {
                    promise.reject(error)
                }
                lock.unlock()
            })
        }
        
        return Future<Array<T>>(promise)
    }
    
    // 返回第一个完成的Future
    public static func race<T>(futures: Array<Future<T>>): Future<T> {
        let promise = Promise<T>()
        
        for (future in futures) {
            future.onSuccess({ value =>
                promise.resolve(value)
            })
            
            future.onFailure({ error =>
                promise.reject(error)
            })
        }
        
        return Future<T>(promise)
    }
    
    // 创建已完成的Future
    public static func completed<T>(value: T): Future<T> {
        let promise = Promise<T>()
        promise.resolve(value)
        return Future<T>(promise)
    }
    
    // 创建已失败的Future
    public static func failed<T>(error: Exception): Future<T> {
        let promise = Promise<T>()
        promise.reject(error)
        return Future<T>(promise)
    }
}

// HTTP响应数据
public struct HttpResponse {
    public let statusCode: Int32
    public let body: String
    public let headers: HashMap<String, String>
}

// 异步HTTP客户端
public class AsyncHttpClient {
    private let scheduler: TaskScheduler
    private let connectionTimeout: Int64 = 5000
    
    public init(poolSize: Int64 = 10) {
        this.scheduler = TaskScheduler(poolSize)
    }
    
    // 异步GET请求
    public func get(url: String): Future<HttpResponse> {
        return scheduler.execute({
            this.performRequest("GET", url, None)
        })
    }
    
    // 异步POST请求
    public func post(url: String, body: String): Future<HttpResponse> {
        return scheduler.execute({
            this.performRequest("POST", url, body)
        })
    }
    
    // 实际执行HTTP请求
    private func performRequest(method: String, 
                                url: String, 
                                body: ?String): HttpResponse {
        // 解析URL
        let urlParts = parseUrl(url)
        let host = urlParts.host
        let port = urlParts.port
        let path = urlParts.path
        
        // 创建套接字连接
        let socket = SocketWrapper.createTcpClient(host, port)
        socket.setTimeout(Int32(connectionTimeout / 1000))
        
        try {
            // 构造请求
            var request = "${method} ${path} HTTP/1.1\r\n"
            request += "Host: ${host}\r\n"
            request += "Connection: close\r\n"
            
            if (body != None) {
                request += "Content-Length: ${body!.size}\r\n"
            }
            
            request += "\r\n"
            
            if (body != None) {
                request += body!
            }
            
            // 发送请求
            socket.sendString(request)
            
            // 接收响应
            let responseData = socket.receiveAll()
            let responseText = String.fromUtf8(responseData)
            
            // 解析响应
            return parseResponse(responseText)
        } finally {
            socket.close()
        }
    }
    
    private func parseUrl(url: String): (host: String, port: Int32, path: String) {
        // 简化实现，实际应使用完整的URL解析
        let withoutProtocol = url.replaceFirst("http://", "")
        let parts = withoutProtocol.split('/')
        let hostPort = parts[0].split(':')
        
        let host = hostPort[0]
        let port = if (hostPort.size > 1) { 
            Int32.parse(hostPort[1]) 
        } else { 
            80 
        }
        let path = "/" + parts[1..].join("/")
        
        return (host, port, path)
    }
    
    private func parseResponse(responseText: String): HttpResponse {
        let lines = responseText.split('\n')
        
        // 解析状态行
        let statusLine = lines[0]
        let statusParts = statusLine.split(' ')
        let statusCode = Int32.parse(statusParts[1])
        
        // 解析headers
        let headers = HashMap<String, String>()
        var bodyStartIndex = 0
        
        for (i in 1..lines.size) {
            let line = lines[Int(i)].trim()
            if (line.isEmpty()) {
                bodyStartIndex = i + 1
                break
            }
            
            let colonIndex = line.indexOf(':')
            if (colonIndex > 0) {
                let key = line.substring(0, colonIndex).trim()
                let value = line.substring(colonIndex + 1).trim()
                headers.put(key, value)
            }
        }
        
        // 提取body
        let bodyLines = lines[bodyStartIndex..]
        let body = bodyLines.join("\n")
        
        return HttpResponse(statusCode, body, headers)
    }
    
    public func shutdown() {
        scheduler.shutdown()
    }
}

// 批量请求管理器
public class BatchRequestManager {
    private let client: AsyncHttpClient
    private let maxConcurrent: Int64
    
    public init(maxConcurrent: Int64 = 10) {
        this.client = AsyncHttpClient(maxConcurrent)
        this.maxConcurrent = maxConcurrent
    }
    
    // 批量并发请求
    public func batchGet(urls: Array<String>): Future<Array<HttpResponse>> {
        let futures = ArrayList<Future<HttpResponse>>()
        
        for (url in urls) {
            futures.append(client.get(url))
        }
        
        return Futures.all(futures.toArray())
    }
    
    // 带重试的请求
    public func getWithRetry(url: String, maxRetries: Int32 = 3): Future<HttpResponse> {
        return retryRequest(url, 0, maxRetries)
    }
    
    private func retryRequest(url: String, 
                             currentAttempt: Int32,
                             maxRetries: Int32): Future<HttpResponse> {
        let promise = Promise<HttpResponse>()
        
        client.get(url).onSuccess({ response =>
            promise.resolve(response)
        }).onFailure({ error =>
            if (currentAttempt < maxRetries) {
                println("Request failed, retrying... (${currentAttempt + 1}/${maxRetries})")
                
                // 递归重试
                let retryFuture = this.retryRequest(url, currentAttempt + 1, maxRetries)
                retryFuture.onSuccess({ response =>
                    promise.resolve(response)
                }).onFailure({ retryError =>
                    promise.reject(retryError)
                })
            } else {
                promise.reject(error)
            }
        })
        
        return Future<HttpResponse>(promise)
    }
}

main(): Int64 {
    println("=== Async HTTP Client Example ===\n")
    
    // 示例1：单个异步请求
    println("1. Single async request:")
    let client = AsyncHttpClient(5)
    
    let future = client.get("http://api.example.com/users/1")
    
    future.onSuccess({ response =>
        println("Status: ${response.statusCode}")
        println("Body: ${response.body}")
    }).onFailure({ error =>
        println("Request failed: ${error.getMessage()}")
    })
    
    // 等待完成
    Thread.sleep(2000)
    println()
    
    // 示例2：链式操作
    println("2. Chained operations:")
    client.get("http://api.example.com/users")
        .map({ response => response.body.size })
        .map({ size => "Response size: ${size} bytes" })
        .onSuccess({ message =>
            println(message)
        })
    
    Thread.sleep(2000)
    println()
    
    // 示例3：并发多个请求
    println("3. Concurrent requests:")
    let urls = [
        "http://api1.example.com/data",
        "http://api2.example.com/data",
        "http://api3.example.com/data"
    ]
    
    let batchManager = BatchRequestManager(10)
    let startTime = System.currentTimeMillis()
    
    let batchFuture = batchManager.batchGet(urls)
    
    batchFuture.onSuccess({ responses =>
        let elapsed = System.currentTimeMillis() - startTime
        println("All requests completed in ${elapsed}ms")
        println("Received ${responses.size} responses")
        
        for (i in 0..responses.size) {
            println("Response ${i + 1}: Status ${responses[Int(i)].statusCode}")
        }
    }).onFailure({ error =>
        println("Batch request failed: ${error.getMessage()}")
    })
    
    Thread.sleep(5000)
    println()
    
    // 示例4：带重试的请求
    println("4. Request with retry:")
    let retryFuture = batchManager.getWithRetry("http://flaky-api.example.com/data")
    
    retryFuture.onSuccess({ response =>
        println("Finally succeeded: ${response.statusCode}")
    }).onFailure({ error =>
        println("All retries failed: ${error.getMessage()}")
    })
    
    Thread.sleep(3000)
    
    // 清理资源
    client.shutdown()
    
    return 0
}

// ✅ 推荐：使用链式操作
client.get(url)
    .map({ response => parseJson(response.body) })
    .flatMap({ data => saveToDatabase(data) })
    .onSuccess({ _ => println("Success!") })
    .onFailure({ error => logError(error) })

// ❌ 避免：嵌套回调地狱
client.get(url).onSuccess({ response =>
    let data = parseJson(response.body)
    saveToDatabase(data).onSuccess({ _ =>
        println("Success!")
    })
})