A simple event bus using RxSwift that make it as short as possible

NBus.swift

//
//  NEvent.swift
//  dailycalcium
//
//  Created by Nam Nguyen on 10/26/18.
//  Copyright © 2018 Impala Intech Limited. All rights reserved.

// NOTE: check comment for usage
//

import Foundation
import RxSwift
import RxCocoa
// Note:
// class require of RxSwift

enum NBusThread {
    case main, background, utility
    func toScheduler() -> SchedulerType {
        switch self {
        case .main:
            return MainScheduler.instance
        case .background:
            return ConcurrentDispatchQueueScheduler(qos: .background)
        case .utility:
            return ConcurrentDispatchQueueScheduler(qos: .utility)
        }
    }
}

class NBus {
    // where bus go on
    private static var station = [String: [NPassenger]]()
    
    // register a method of target to observer
    static func register(passenger: AnyObject, bus eventTag: String, callAt callThread: NBusThread, _ handler: @escaping (AnyObject, [Any]?) -> ()) {
        // create new sit for this bus
        let newPassenger = NPassenger()
        newPassenger.id = UInt(bitPattern: ObjectIdentifier(passenger))
        newPassenger.busId = eventTag
        newPassenger.disposable = newPassenger.obserable
            .observeOn(callThread.toScheduler())
            .subscribe(onNext: { (obj) in handler(passenger, obj) })
        
        // check if we already add this buss
        if var passengers = station[eventTag] {
            var shouldAdd = true
            for oldPassenger in passengers {
                // there is no glich on matrix that 2 version of passenger on bus
                // at same time
                if oldPassenger.id == newPassenger.id {
                    shouldAdd = false
                    break
                }
            }
            if shouldAdd {
                passengers.append(newPassenger)
            }
        }else{
            station[eventTag] = [newPassenger]
        }
    }
    
    // clean up all passenger sit for this passenger on all the bus
    static func unregister(passenger: AnyObject) {
        let id = UInt(bitPattern: ObjectIdentifier(passenger))
        for busId in station.keys {
            var passengers = station[busId]!
            for i in 0..<passengers.count {
                if passengers[i].id == id {
                    passengers[i].disposable!.dispose()
                    passengers.remove(at: i)
                    break
                }
            }
            if passengers.count == 0 {
                station.removeValue(forKey: busId)
            }
        }
    }
    
    // clean up all passenger on the bus
    static func unregister(bus: String) {
        if var passengers = station[bus] {
            for i in 0..<passengers.count {
                passengers[i].disposable!.dispose()
            }
            passengers.removeAll()
            station.removeValue(forKey: bus)
        }else{
            print("[NBus] unregister do not found: \(bus)")
        }
    }
    
    // when passenger want to take the bus just call it
    static func call(bus eventTag: String,_ data: [Any]? = nil) {
        if let passengers = station[eventTag] {
            passengers.forEach { (passenger) in
                passenger.subject.onNext(data)
            }
        }else{
            print("[NBus] do not found: \(eventTag) please make sure it was registered.")
        }
    }
    
    // just a sit for any passenger and only one
    class NPassenger {
        let subject = PublishSubject<[Any]?>()
        var disposable : Disposable? = nil
        var obserable: Observable<[Any]?> { get { return subject } }
        var busId: String = ""
        var id: UInt = 0
    }
}

extension NBus {
    
    // call a method on a thread
    static func run(callAt thread: NBusThread,_ callback: @escaping () -> ()) {
        run(callAt: thread, delay: 0.0, callback)
    }
    
    // call a method on a thread
    static func run(callAt thread: NBusThread,delay: Double,_ callback: @escaping () -> ()) {
        run(doAt: thread, callAt: thread, delay: delay, callback)
    }
    
    // call a method in details
    static func run(doAt doThread: NBusThread, callAt callThread: NBusThread, delay: Double, _ callback: @escaping ()-> ()) {
        Observable<Int>.timer(delay, scheduler: doThread.toScheduler()).take(1)
            .subscribeOn(doThread.toScheduler())
            .observeOn(callThread.toScheduler())
            .subscribe(onNext: { (_) in
                callback()
            })
    }
}

Using

1, register event

// should call on viewWillAppear()
NBus.register(passenger: self,  bus: "BUS_ID_CONST_SOME_WHERE", callAt: .main) { (passenger, data) in 
    // current object : passenger -> no need to pass [weak self]
    print("Event called with data: \(data?.count)")
    if let _data = data {
        for e in data { print(e) }
    }
    passenger.didMove(...)
}

2, call event

    NBus.call(bus: "BUS_ID_CONST_SOME_WHERE", ["string", 10, 3.14, ["AAA", "BBB"]])

result should show:

   Event called with data: Optional(4)
   string
   10
   3.14
   ["AAA", "BBB"]

3, unregister

    // call this for clean up event registered by current object and should be called on viewWillDisappear
    NBus.unregister(passenger: self) // remove all event register from self object
    Nbus.unregister(busId: "BUS_ID_CONST_SOME_WHERE") // remove all event with Id

4, some bonus method

    // call after delay amount of time
    NBus.run(callAt: .main, delay: 3.0) {
        // do something in main thread after 3 seconds
    }
    // call on a thread next frame ?
    NBus.run(callAt: .main) {}