/*
 * @Copyright (c) Huawei Technologies Co., Ltd. 2023-2024. All rights reserved.
 */
package mqtt4cj.mqttv3

class WebSocketReceiver {
    private static let CLASS_NAME: String = "WebSocketReceiver"
    private let lifecycle = Mutex()
    private let log: IMqttLogger = LoggerFactory.getLogger(MQTT_CLIENT_MSG_CAT, CLASS_NAME)
    private let websocket: WebSocket
    private let pipedStream: IOStream
    private var running: Bool = false
    private var stopping: Bool = false
    private var receiverFuture: ?Future<Unit> = None
    var threadName: ?String = None

    public init(websocket: WebSocket, pipedStream: IOStream) {
        this.websocket = websocket
        this.pipedStream = pipedStream
    }

    /**
     * Starts up the WebSocketReceiver's thread
     *
     * @param threadName The name of the thread
     */
    public func start(threadName: String): Unit {
        this.threadName = threadName
        let methodName: String = "start"

        // @TRACE 855=starting
        log.trace(CLASS_NAME, methodName, "855")
        synchronized(lifecycle) {
            if (!running) {
                running = true
                let future = spawn {
                    this.run()
                }
                receiverFuture = future
            }
        }
    }

    /**
     * Stops this WebSocketReceiver's thread.
     * This call will block.
     */
    public func stop(receiver!: Bool = false): Unit {
        let methodName: String = "stop"
        this.stopping = true
        var closed: Bool = false
        synchronized(lifecycle) {
            // @TRACE 850=stopping
            log.trace(CLASS_NAME, methodName, "850")
            if (running) {
                this.running = false
                closed = true
                closeOutputStream()
            }
        }
        if (closed && !receiver && receiverFuture.isSome()) {
            // Wait for the thread to finish
            // This must not happen in the synchronized block, otherwise we can deadlock ourselves!
            if (receiverFuture().thread != Thread.currentThread) {
                receiverFuture?.cancel()
                try {
                    receiverFuture?.get(SHUTDOWN_TIMEOUT)
                }catch (_) {
                    // ignore
                }
            }
        }
        receiverFuture = None

        // @TRACE 851=stopped
        log.trace(CLASS_NAME, methodName, "851")
    }

    public func run(): Unit {
        Thread.currentThread.name = "WebSocketReceiver.run"
        let methodName: String = "run"
        while (running) {
            try {
                // @TRACE 852=network read message
                log.trace(CLASS_NAME, methodName, "852")
                var frame = websocket.read()
                while (true) {
                    // only BinaryWebFrame(opcode=0x2) and CloseWebFrame(opcode=0x8)
                    match (frame.frameType) {
                        case BinaryWebFrame =>
                            pipedStream.write(frame.payload)
                            if (frame.fin) {
                                pipedStream.flush()
                                break
                            }
                        case CloseWebFrame =>
                            if (!stopping) {
                                throw IllegalStateException("Server sent a WebSocket Frame with the Stop OpCode")
                            }
                            break
                        case v => throw IllegalStateException("Invalid Frame: Opcode: ${v}")
                    }
                    frame = websocket.read()
                }
            } catch (e: IOException | IllegalStateException | SocketException) {
                
                log.trace(CLASS_NAME, methodName, "${e}")
                // Exception occurred whilst reading the stream.
                this.stop(receiver: true)
            }
        }
    }

    private func closeOutputStream(): Unit {
        (pipedStream as Resource)?.close()
    }
}