package com.clinicpatientqueueexample.messaging;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

public abstract class AbstractBroadcaster {

    private final ConcurrentMap<String, ConcurrentLinkedQueue<Consumer<String>>> listeners = new ConcurrentHashMap<>();
    private final ExecutorService executorService = Executors.newSingleThreadExecutor();

    public void register(String topic, Consumer<String> listener) {
        ConcurrentLinkedQueue<Consumer<String>> topicListeners = listeners.putIfAbsent(topic, new ConcurrentLinkedQueue<>());
        if (topicListeners == null) {
            topicListeners = listeners.get(topic);
        }
        topicListeners.add(listener);
    }

    public void unregister(String topic, Consumer<String> listener) {
        final ConcurrentLinkedQueue<Consumer<String>> topicListeners = listeners.get(topic);
        if (topicListeners != null) {
            topicListeners.remove(listener);
        }
    }

    public void broadcast(String topic, String message) {
        final ConcurrentLinkedQueue<Consumer<String>> topicListeners = listeners.get(topic);
        if (topicListeners != null) {
            for (final Consumer<String> listener : topicListeners) {
                executorService.execute(() -> listener.accept(message));
            }
        }
    }

}

@@ -1080,33 +1080,33 @@
 (topicListeners 
-!
+=
 = null) %7B%0A      
@@ -1095,32 +1095,58 @@
  == null) %7B%0A    
+        return;%0A        %7D%0A
         topicLis
@@ -1166,24 +1166,101 @@
 (listener);%0A
+        if (topicListeners.isEmpty()) %7B%0A            listeners.remove(topic);%0A
         %7D%0A  

Remove listeners object from broadcaster if topic is empty