Java Pub Sub implementation using broker

Question

I have developed a simple pubsub event subscription in java. Something similar to this thread

Simple Pub-Sub Event Emitter

These are the goals.

1. Class should be able to "listen" to events happening in the module.
2. It should be disconnected from the event emitter. That is, the listeners shouldn't have to know which classes may be emitting events.
3. All the subscribers should be notified parallelly.

My Broker.java

import java.util.*;

public final class  Broker {
    private Object mutex = new Object();
    static class CommonTopics{
        private CommonTopics(){}
        public static final String ON_CACHE_RESET = "onCacheReset";
    }
    private static Broker brokerInstance;
    private Broker(){

    }
    public static Broker getInstance(){
        if(null== brokerInstance){
            brokerInstance = new Broker();
        }
        return brokerInstance;
    }

    private Map<String, Set<Subscriber>> subscribers = new HashMap<>();

    public boolean deregister(String topic, Subscriber subscriber) {
        synchronized (mutex) {
            final Set<Subscriber> subs = this.subscribers.get(topic);
            return subs.remove(subscriber);
        }
    }

    public boolean register(String topic, Subscriber subscriber) {
        boolean returnVal;
        synchronized (mutex) {
            if (subscribers.containsKey(topic)) {
                returnVal = subscribers.get(topic).add(subscriber);
            } else {
                Set<Subscriber> sub = new HashSet<>();
                returnVal = sub.add(subscriber);
                subscribers.put(topic, sub);
            }
        }
        return returnVal;
    }

    public void sendMessage(String topic,Map map){
        synchronized (mutex) {
            final Set<Subscriber> sub = this.subscribers.get(topic);
            sub.parallelStream().forEach(subscriber -> subscriber.update(map));
        }
    }

}

My Subscriber interface

import java.util.Map;

public interface Subscriber {
    public void update(Map map);
}

And an example subscriber

import java.util.Map;

public class ExampleSubscriber implements Subscriber {
    @Override
    public void update(Map map) {
        System.out.println(map.get("oldCachedValue"));
        System.out.println(map.get("newCachedValue"));
    }

    public static void main(String[] args) {
        ExampleSubscriber subscriber = new ExampleSubscriber();
        Broker.getInstance().register(Broker.CommonTopics.ON_CACHE_RESET,subscriber);
    }
}

Example event emmitter

Map m = new HashMap();
        m.put("oldCachedValue","Yes");
        m.put("newCachedValue","No");
        Broker.getInstance().sendMessage(Broker.CommonTopics.ON_CACHE_RESET,m);

Answer 1

• First of all, mutex should be final, since you are synchronizing on it. In the current state of your code, it wouldn't make much difference, since it is private anyway and never exposed outside the class, but still, if you make it final, then the field mutex is protected against modification by the compiler, reducing the risk of bugs.

• If you don't ever intend to change brokerInstance, i.e. if Broker is intended to be a singleton, you can also make brokerInstance final and initialize it immediately with its declaration. This would eliminate the need for the if construct in the method getInstance(). After googling about this, I found out that the purpose of what you are doing is probably lazy initialization. However, in this question on stackoverflow, some points are made in favor of the approach with the final variable (especially the point about thread safety might be interesting to you).

• You might consider taking advantage of the class ConcurrentHashMap for your synchronization purposes. A ConcurrentHashMap is optimized for synchronized access in that it is not locked entirely during an operation, so that operations that do not conflict with each other can be performed simultaneously (you might want to take a look at this).

  For example, you could make each Set<Subscriber> in Broker.subscribers a ConcurrentHashMap.KeySetView (which serves as the Set equivalent of a ConcurrentHashMap, because there is no class ConcurrentHashSet) instead of a regular HashMap. This would at least make the synchronized block in Broker.deregister(String, Subscriber) unnecessary. Making Broker.subscribers itself a ConcurrentHashMap might also help, but you still have to be careful, for example in the method Broker.register(String, Subscriber), because, without any other synchronization, the first call to subscribers.containsKey(topic) might yield false, so the else block is entered, but before the call subscribers.put(topic, sub) is reached, subscribers might have been changed and could now already contain a mapping for topic which would then be overwritten. There seem to be special methods in the class ConcurrentHashMap for such cases (like the compute methods), but since this is merely a general suggestion, I won't go into detail about it.

• Your code lacks handling of exceptional cases, for instance, if the methods deregister(String, Subscriber) or sendMessage(String, Map) are called with a topic that is not contained in the key set of subscribers.

• As I already hinted at in a comment, using raw types is a dangerous undertaking. It is not wrong per se, but the risk of bugs is higher because generics provide compile-time type-safety which you are not taking advantage of if you use raw types. I don't know what your code might do with the map, so maybe it won't be a problem, but you might still take a look at this or this.

• Also note that the map that is being passed to Broker.sendMessage(String, Map) will be accessed concurrently in sub.parallelStream().forEach(subscriber -> subscriber.update(map)), which might not be desirable if the map can be modified by the subscribers.