path: root/Net1090/NetADSBDecoder.swift

//
//  NetADSBDecoder.swift
//  Net1090
//
//  Created by Jacky Jack on 19/07/2024.
//

import Foundation
import Network
import NIO

protocol ADSBQueueDelegate: AnyObject {
    //var message_array:Array<String> {get set}
    func push(_ message: String)
    func pop() -> String?
}

class NetADSBHandlder: ChannelInboundHandler {
    typealias InboundIn = ByteBuffer
    typealias OutboundOut = ByteBuffer
    
    var messageDelegate: ADSBQueueDelegate?
    
    func channelActive(context: ChannelHandlerContext) {
        print("Channel is active")
    }
    
    func channelInactive(context: ChannelHandlerContext) {
        print("Channel inactive")
        context.fireChannelInactive()
    }
    
    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
        var buffer = unwrapInboundIn(data)
        let readableBytes = buffer.readableBytes
        if let received =  buffer.readString(length: readableBytes) {
            //print(received,terminator: "")
            /*
            if let message_queue = message_queue {
                message_queue.append(String(received))
                print("Add new messsge")
            }*/
            //if self.message_array != nil {
            //    message_array?.append(String(received))
            //    print("\(message_array?.count)")
            //}
            if self.messageDelegate != nil {
                if received.count == 17 {
                    let trimmed = received.trimmingCharacters(in: .newlines).lowercased()
                    //messageDelegate?.message_array.append(String(trimmed))
                    messageDelegate?.push(String(trimmed))
                    
                } else if received.count == 31  {
                    let trimmed = received.trimmingCharacters(in: .newlines).lowercased()
                    messageDelegate?.push(String(trimmed))
                } else {
                    for line in received.components(separatedBy: .newlines) {
                        let trimmed = line.trimmingCharacters(in: .newlines).lowercased()
                        messageDelegate?.push(String(trimmed))
                    }
                }
            }
        }
    }
    
    func errorCaught(context: ChannelHandlerContext, error: any Error) {
        print("error: \(error.localizedDescription)")
        context.close(promise: nil)
    }
}

final class ADSBQueue: ADSBQueueDelegate {
    private var message_array:Array<String> = []
    private let lock =  NSLock()
    
    func push(_ message: String) {
        lock.lock()
        message_array.append(message)
        lock.unlock()
    }
    
    func pop() -> String? {
        lock.lock()
        defer { lock.unlock() }
        guard !message_array.isEmpty else { return nil }
        return message_array.removeFirst()
    }
    
    var count: Int {
        lock.lock()
        defer { lock.unlock() }
        return message_array.count
    }
}

enum ChannelState: CustomStringConvertible {
    case connected
    case disconnected
    case retry(seconds: Int)
    
    var description: String {
        switch self {
        case .connected: return "connected"
        case .disconnected: return "disconnected"
        case .retry(let seconds): return "connection last retrying in \(seconds) sec"
        }
    }
}

class NetADSBDecoder {
    let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
    var host: String
    var port: Int
    
    var msgarray = ADSBQueue()
    var handler = NetADSBHandlder()
    //var msg_array:Array<String> = []
    
    var onStateChange: ((ChannelState) -> Void)?
    private let stateLock = NSLock()
    private var stopped = false
    private var retries = 0
    private var retrySemaphore = DispatchSemaphore(value: 0)
    private static let retryDelay = 30
    
    private let pathMonitor = NWPathMonitor()
    private let pathLock = NSLock()
    private var currentChannel:Channel? = nil
    
    private var isStopped: Bool {
        stateLock.lock()
        defer {stateLock.unlock()}
        return stopped
    }
    
    
    init(host: String, port: Int) {
        self.host = host
        self.port = port
        //regiter delegate
        handler.messageDelegate = msgarray
    }
    
    func start() throws {
        startPathMonitor()
        defer { pathMonitor.cancel() }
        
        while (!isStopped) {
            setState(.disconnected)
            do {
                let channel = try ClientBootstrap(group: group)
                    .connectTimeout(.seconds(10))
                    .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET),SO_REUSEADDR), value: 1)
                    .channelOption(ChannelOptions.socketOption(.so_keepalive), value: 1)
                    .channelOption(ChannelOptions.socket(SocketOptionLevel(IPPROTO_TCP),TCP_KEEPALIVE), value: 15)
                    .channelInitializer{ channel in
                        channel.pipeline.addHandlers([self.handler])
                    }
                    .connect(host: self.host, port: self.port)
                    .wait()
                setCurrentChannel(channel)
                retries = 0
                setState(.connected)
                try channel.closeFuture.wait()
                clearCurrentChannel()
                print("Connection lost")
            } catch let error {
                clearCurrentChannel()
                print("Connection error \(error)")
            }
            if isStopped {
                break
            }
            retries += 1
            let delay = min(Self.retryDelay, 1 << min(retries, 5))
            setState(.retry(seconds: delay))
            if retrySemaphore.wait(timeout: .now() + .seconds(delay)) == .success, isStopped { break }
        }
        print("Exit connection loop")
    }
    
    
    
    func stop() {
        stateLock.lock()
        stopped = true
        stateLock.unlock()
        _ = retrySemaphore.signal()
        do {
            try group.syncShutdownGracefully()
        } catch {
            print("Error shutting down \(error.localizedDescription)")
        }
        print("Connection closed")
    }
    
    private func setState(_ state: ChannelState) {
        print("socket: \(state)")
        onStateChange?(state)
    }
    
    private func startPathMonitor() {
        /*pathMonitor.pathUpdateHandler = { [weak self] path in
            guard let self else {return}
            print("network connection changed \(path.status)")
            self.closeCurrentChannel()
            if path.status == .satisfied {
                self.retrySemaphore.signal()
            }
        }*/
        pathMonitor.pathUpdateHandler = { path in
            print("network connection changed \(path.status)")
            self.closeCurrentChannel()
            if path.status == .satisfied {
                self.retrySemaphore.signal()
            }
        }
        let queue = DispatchQueue(label: "net1090.connection-monitor")
        pathMonitor.start(queue: queue)
    }
    
    private func setCurrentChannel(_ channel: Channel?) {
        pathLock.lock()
        currentChannel = channel
        pathLock.unlock()
    }
    
    private func clearCurrentChannel() {
        setCurrentChannel(nil)
    }
    
    private func closeCurrentChannel() {
        pathLock.lock()
        let channel = currentChannel
        pathLock.unlock()
        channel?.close(promise: nil)
    }
}