Untitled

/******************************************************************************

Welcome to GDB Online.
  GDB online is an online compiler and debugger tool for C, C++, Python, PHP, Ruby, 
  C#, OCaml, VB, Perl, Swift, Prolog, Javascript, Pascal, COBOL, HTML, CSS, JS
  Code, Compile, Run and Debug online from anywhere in world.

*******************************************************************************/
#include <bits/stdc++.h>
using namespace std;
class Topic;
class Consumer;
class Consumer{
    public : 
    int consumerId;
    void consume(string message){
        cout<<"Consumer "<<consumerId<<" consumed the message : "<<message<<endl;
    }
};
class Topic{
    public : 
    int topicId;
    vector<Consumer*> consumers;
    std::mutex consumersMutex;
    unordered_map<int,queue<message>> queuePerConsumer;
    
    Topic(int topicId) : topicId(topicId){}
    
    void addMessage(string message){
        notifyConsumers(message);
    }
    
    void notifyConsumers(string message){
        vector<std::thread> consumerThreads;
        {
            std::lock_guard<mutex> consumersLock(consumersMutex);
            for(auto consumer : consumers){
                consumerThreads.emplace_back(&Consumer::consume, consumer, message);
            }
        }
        for(auto &consumerThread:consumerThreads){
            consumerThread.join()
        }
    }
    
    void addConsumer(Consumer* consumer){
        std::lock_guard<mutex> consumersLock(consumersMutex);
        consumers.push_back(consumer);
    }
    
};
class Pubsub{
    public : 
    map<int,Topic*> topics;
    std::mutex topicsMutex;
    Topic * createTopic(int topicId){
        std::lock_guard<mutex> topicsLock(topicsMutex);
        Topic * newTopic =  new Topic(topicId);
        topics[newTopic->topicId] = newTopic;
        return newTopic;
    }
    
    void subscribe(Consumer *consumer, int topicId){
        std::lock_guard<mutex> topicsLock(topicsMutex);
        if(!topics.count(topicId)){
            cout<<"Invalid topic with topic id : "<<topicId<<endl;
            return;
        }
        topics[topicId]->addConsumer(consumer);
    }
    void publish(string message, int topicId){
        std::lock_guard<mutex> topicsLock(topicsMutex);
        if(!topics.count(topicId)){
            cout<<"Invalid topic with topic id : "<<topicId<<endl;
            return;
        }
        topics[topicId]->addMessage(message);
    }
    
};




int main() {
    Consumer * c1 = new Consumer(1);
    Consumer * c2 = new Consumer(2);
    Consumer * c3 = new Consumer(3);
    
    Pubsub pb;
    pb.createTopic(1);
    pb.subscribe(c1,1);
    pb.subscribe(c2,1);
    
    pb.publish("I am great",1);
    pb.subscribe(c3,1);
    pb.publish("I am lalloo",1);
}