Broker.java

import java.util.*;

public final class Broker {
    private final Object mutex = new Object();

    static class CommonTopics {
        private CommonTopics() {
        }

        public static final String ON_TAX_CHANGE = "onTaxChange";

    }

    private static final Broker instance = new Broker();

    private Broker() {

    }

    public static Broker getInstance() {
        return instance;
    }

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

    public boolean deregister(String topic, Subscriber subscriber) {
        if (!subscribers.containsKey(topic)) {
            return false;
        }
        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) {
        if (!subscribers.containsKey(topic)) {
            return;
        }
        synchronized (mutex) {
            final Set<Subscriber> sub = this.subscribers.get(topic);
            sub.parallelStream().forEach(subscriber -> {
                        Map immutableMap = Collections.unmodifiableMap(new HashMap<>(map));
                        Context ctx = new Context();
                        ctx.data = immutableMap;
                        ctx.topic = topic;
                        subscriber.update(ctx);
                    }
            );
        }
    }
    
    public class Context{
    	private String topic;
    	private Map data;
		public String getTopic() {
			return topic;
		}
		public Map getData() {
			return data;
		}
    	
    }

}