Lightweight flexible observer framework using weak references I found a need for a simple, generic publish-subscribe framework for my Java application.  I looked over a few implementations on Code Review and I think mine avoids most of the pitfalls.  Single-class implementation using a Consumer to receive messages and an Object for indexing.

import java.util.Collections;
import java.util.Map;
import java.util.WeakHashMap;
import java.util.function.Consumer;

public class Broadcaster&lt;M&gt; {
    private final Map&lt;Object,Consumer&lt;M&gt;&gt; listeners = Collections.synchronizedMap(new WeakHashMap&lt;&gt;());

    public void post(M message){
        listeners.values().forEach(c-&gt;c.accept(message));
    }

    public void listen(Object listener, Consumer&lt;M&gt; consumer){
        listeners.put(listener, consumer);
    }

    public void ignore(Object listener){
        listeners.remove(listener);
    }
}


Here's a demo of (most of) the functionality.  Hopefully I wrote enough to give a general idea of how it works.

import java.util.function.Consumer;

public class NeedsToBroadcast{
    private Broadcaster&lt;Thing&gt; thingBroadcaster = new Broadcaster&lt;&gt;();

    public void listen(Object listener, Consumer&lt;Thing&gt; consumer){
        thingBroadcaster.listen(listener, consumer);
    }

    public void ignore(Object listener){
        thingBroadcaster.ignore(listener);
    }

    public void doStuff(int i){
        do{
            Thing thing = StuffProcessor.process(i);
            if(thing != null){
                // Multiple objects care about the Things we get
                i = thing.getNumber();
                thingBroadcaster.post(thing);
            }else{
                i = 0;
            }
        }while(i&gt;1);
    }

    public static void main(String... args){
        NeedsToBroadcast ntb = new NeedsToBroadcast();
        StandingLog standingLog = new StandingLog(ntb);

        ntb.doStuff(20);

        System.out.println(standingLog.toString());
    }

    public static class Thing{
        private final int number;
        private final String text;

        private Thing(int number, String text){
            this.number = number;
            this.text = text;
        }

        public int getNumber(){
            return number;
        }

        public String getText(){
            return text;
        }
    }

    public static class StandingLog{
        private final StringBuffer buffer = new StringBuffer();

        public StandingLog(NeedsToBroadcast ntb){
            // As long as this object has a reference, it will receive updates
            ntb.listen(this, (thing)-&gt;{
                buffer.append(thing.getNumber()).append(":").append(thing.getText()).append('\n');
            });
        }

        @Override
        public String toString(){
            return buffer.toString();
        }
    }
}


Update

I'm not trying to make something competitive with, say, 0MQ.  I'm trying to create a drop-in, single-class replacement for something like Java's Observable and Observer, with the added benefits of thread safety, cleanup for GC-elegible objects, and generic typing.

The original title of this post used "publish-subscribe".  I changed it to "observer" since, while I feel they are similar enough for this class, the distinction seems to have given people the wrong idea.

It seems the demo I posted doesn't do a fantastic job of demonstrating the functionality.  I'll see if I can rework something for clarity's sake.
I have some "readability-issues". It usually helps to name your methods with a verb and a noun, so, post would be postMessage, for instance.

listen and ignore are not named correctly, in my opinion. If I call listen(listener, consumer), I do not know what this method will do with the two parameters, it feels like, the method adds the listener to the consumer. I also have no idea, that the listener is used for unsubscribing later. If I call ignore(listener), it feels like I pause the listener, and can activate it later. addListener and removeListener would have been more clear.

Also, since we're talking about publishers and subscribers, I'd recommend to use the terminology of this pattern. Publisher, Subscriber (well, Consumer is given by the jdk), publish, subscribe and unsubscribe.

Well, it's a bit smart-arse-y, but, since I'm here,... By definition, or at least by the definition I know/remember, one thing the publish/subscribe pattern wants to achieve is, that the subscribers have the ability to subscribe to specific messages types, e.g. : You subscribe to a news feed, but only the sports part. So, I recommend to rework the terminology, because a user of your API can have different expectations of the API. (Yes, I'm one of them :P)



StuffProcessor is missing, btw, so I can't run the code.



I had a really hard time to understand your sample code. I tried to make an easier example of the usage:

    public class BroadcasterSampleUsage {
                public static void main(String[] args) {
                           Broadcaster&lt;Message&gt; broadcaster = new Broadcaster&lt;&gt;();
                           Object someIdentifier = new Object();

                           broadcaster.listen(someIdentifier, new MessageConsumer());

                           broadcaster.post(new Message("helloes"));
                }

                private static class MessageConsumer implements Consumer&lt;Message&gt; {
                           @Override
                           public void accept(Message t) {
                                       System.out.println("Received message: " + t.getMessage());
                           }
                }

                private static class Message {
                           private final String message;

                           public Message(String message) {
                                       this.message = message;
                           }

                           public String getMessage() {
                                       return message;
                           }
                }
    }


I'm not 100% certain if that's the intention of your API, after I've seen the usage code, but alas, if I understood your intention correctly, that would be easier to read.



The only Reason for the Object listener is to have it as key object, right? Now, why is that confusing for me: You call that thing listener. What is it listening for? And what's the difference between listening and consuming? 

In your sample code, you actually do use this as key and use a lambda expression as parameter. So what if I would actually would overwrite the accept method in the StandingLog type, change the method signature in Broadcaster from

public void listen(Object listener, Consumer&lt;M&gt; consumer)


to

public void listen(Consumer&lt;M&gt; consumer)


, and instead of putting the listener and the consumer to the map, I use the consumer as key and value, wouldn't I achieve the same thing? If yes, I think you should be able to use this one: http://docs.oracle.com/javase/8/docs/api/java/util/Collections.html#newSetFromMap(java.util.Map).



Also, check the JavaDoc about the synchronized* methods. Synchronization may not be done yet: It is imperative that the user manually synchronize on the returned map when iterating over any of its collection views. I do think, any means keys and values.



Hope this helps,

slowy