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<M> {
private final Map<Object,Consumer<M>> listeners = Collections.synchronizedMap(new WeakHashMap<>());
public void post(M message){
listeners.values().forEach(c->c.accept(message));
}
public void listen(Object listener, Consumer<M> 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<Thing> thingBroadcaster = new Broadcaster<>();
public void listen(Object listener, Consumer<Thing> 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>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)->{
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.
