Pusher mass subscribe

Pusher mass subscribe.swift

// Main subscription method
private func subscribeToRooms(withChats chats: Chats, completion: @escaping (_ conversations: ChatsWithSubscriber) -> Void) {
        guard let user = chatUser else {return}
        
        let activeCurrentUserChats = chats.filter { (aChat) -> Bool in
            return user.rooms.contains(where: { (pcRoom) -> Bool in
                if aChat is ProjectChat && pcRoom.lastMessageAt == nil {
                    return false
                }
                return aChat.room.roomID == pcRoom.id
            })
        }
        
        let queue = OperationQueue()
        queue.isSuspended = true
        
        let group = DispatchGroup()
        
        
        let syncedValidChats = PCSynchronizedArray<Chat>()
        var newlySubscribedChats: ChatsWithSubscriber = []
        
        syncedValidChats.append(with: activeCurrentUserChats) {
            for i in 0..<syncedValidChats.count {
                group.enter()
                
                let chat = syncedValidChats[i]
                
                let op = PusherSubscriptionOperation(chat: chat, user: user, completion: { (sub) in
                    newlySubscribedChats.append(ChatWithSubscriber(chat: chat, chatRoomSubscriber: sub))
                    group.leave()
                })
                
                queue.addOperation(op)
            }
            
            group.notify(queue: .main) {
                completion(newlySubscribedChats)
            }
            
            queue.isSuspended = false
        }
}

fileprivate class PusherSubscriptionOperation: AsynchronousOperation {
    var completion: ((_ sub: PusherChatRoomSubscriber)->Void)
    private var chat: Chat
    private var user: PCCurrentUser

    init(chat: Chat, user: PCCurrentUser, completion: @escaping (_ sub: PusherChatRoomSubscriber) -> Void) {
        self.completion = completion
        self.chat = chat
        self.user = user
        super.init()
    }
    
    override func main() {
        super.main()
        let chatRoomSubscriber = PusherChatRoomSubscriber(withChatRoom: chat.room, forUser: user)
        chatRoomSubscriber?.subscribeToRoom() { [weak self] in
            guard let chatRoomSubscriber = chatRoomSubscriber else {fatalError()}
            self?.completion(chatRoomSubscriber)
            self?.state = .finished
        }
        
    }
}


extension PCSynchronizedArray {
    public func append(with collection: Array<T>, completion: @escaping () -> Void) {
        let group = DispatchGroup()
        
        collection.forEach({ _ in
            group.enter()
        })
        
        group.notify(queue: .main) {
            completion()
        }

        collection.forEach({
            append($0, completionHandler: {
                group.leave()
            })
        })
    }
}

  // The subscribe function on our Model wrapper PusherChatRoomSubscriber
public func subscribeToRoom(completion: @escaping () -> Void) {
	pusherUser.subscribeToRoomMultipart(room: pusherRoom, roomDelegate: self) { [weak self] error in
	    guard let self = self else {fatalError()}
	    if let error = error {
		Logger.shared.log("Couldn't subscribe to room: \(self.pusherRoom) \(error.localizedDescription)", topics: .pusher)
	    }
	    completion()
	}
}