SocketIO + RxSwift = Awesomeness 🚀

MySocketManager.swift

import RxCocoa
import RxSwift
import SocketIO


final class MySocketManager {
    struct Relays {
        // Custom events:

        /// User properties updated
        let user = PublishRelay<[Any]>()

        /// User activities added
        let activities = PublishRelay<[Any]>()

        // Socket events:

        /// Listen for socket connection changes.
        /// Relays `true` on `connected`, `false` on `disconnected`.
        let socketConnection = PublishRelay<Bool>()
    }

    static let shared = MySocketManager()

    private var socketManager: SocketManager!
    private let disposeBag = DisposeBag()
    private var socket: SocketIOClient {
        return socketManager.socket(forNamespace: "/my-namespace")
    }

    // Publicly accessible relays
    let relays = Relays()

    // Call start from the singleton
    func start() {

        guard socketManger == nil else {
            return
        }

        // Initialize socket manager
        socketManager = SocketManager(
            socketURL: MyConfig.currentEnvironment.socketURL,
            config: [
                .compress,
                // authentication headers or cookies.
                .extraHeaders(AuthManager.shared.authHeaders)
            ]
        )

        addListeners()
        connect()
    }

    func addListeners() {
        socket.listen(MysocketEvents.userUpdates.rawValue, relay: relays.user)
        socket.listen(MysocketEvents.userActivities.rawValue, relay: relays.user)

        socket.listen(SocketClientEvent.connect.rawValue, result: true, relay: relays.socketConnection)
        socket.listen(SocketClientEvent.disconnect.rawValue, result: false, relay: relays.socketConnection)
    }

    func connect() {
        socket.connect()
    }
}

SocketIO+PublishRelay.swift

import RxSwift
import RxCocoa
import SocketIO

extension SocketIOClient {
    func listen(event: String, relay: PublishRelay<[Any]>) {
        on(event) { items, _ in
           relay.accept(items)
        }
    }
    
    func listen<ResultType>(event: String, result: ResultType, relay: PublishRelay<ResultType>) {
        on(event) { _, _ in
           relay.accept(result)
        }
    }
}

Usage.swift

// React to user updates:

MySocketManager.relays.user
  .bind { (updates: [Any]) in
       Single
         .just(updates)
         .observeOn(BackgroundScheduler.instance)
         .decode()  // An Rx extension that decodes [Any] to `User(profile:_, account:_)` structure. 
                    // HMU if interested in learning about this extension
         .observeOn(MainScheduler.instance)
         .map { (user: User) in
             // Update UI to reflect updated user fieds.
         }
         .subscribe()
         .disposed(by: self.disposeBag)
  }
  .subscribe()
  .disposed(by: disposeBag)


// React to Activities:
MySocketManager.relays.activities
  .bind { (activities: [Any]) in
       Single
         .just(activities)
         .observeOn(BackgroundScheduler.instance)
         .decode()
         .observeOn(MainScheduler.instance)
         .map { (activity: Activity) in
             // Update UI to show new activity.
         }
         .subscribe()
         .disposed(by: self.disposeBag)
  }
  .subscribe()
  .disposed(by: disposeBag)