1
\$\begingroup\$

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.

\$\endgroup\$

2 Answers 2

1
\$\begingroup\$

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<Message> broadcaster = new Broadcaster<>();
                           Object someIdentifier = new Object();

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

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

                private static class MessageConsumer implements Consumer<Message> {
                           @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<M> consumer)

to

public void listen(Consumer<M> 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

\$\endgroup\$
2
  • \$\begingroup\$ Thanks a ton for the feedback. I'll probably do some renaming. The reason for object keys is because I'm using a WeakHashMap, which removes entries if they're garbage-collected. That's partially why the code was complicated; I tried to demonstrate a use case for this. \$\endgroup\$
    – ndm13
    Aug 30, 2017 at 14:12
  • \$\begingroup\$ Collections.synchronizedMap(...) synchronizes all the calls I make (values and put), and I only make one call per method. Therefore, external synchronization is just overhead. \$\endgroup\$
    – ndm13
    Aug 30, 2017 at 14:15
0
\$\begingroup\$

I guess I'm willing to entertain the notion that this class might be competitive. I would be more convinced if I saw an app ported to this and to say, 0mq and to kafka.

This implementation seems to tackle the single-host case, which is maybe not very interesting if a pub-sub system is being employed for scaling. Exploiting multiple CPU cores even on a single host may be an issue for this code -- I see no unit tests that stress that use case.

When I looked for specific items that you solicited constructive criticism on, all I noticed was "avoids most of the pitfalls". I would welcome more specific claims about pitfalls and approaches to avoiding them. Just to offer a straw man, I'll posit that 0mq is a strong contender, and evaluating how a a particular app { uses your framework, or uses the 0mq framework } would offer an interesting comparison metric. There may be something here, and I'd love to see the best code demonstrate the lowest latency!

\$\endgroup\$
1
  • 1
    \$\begingroup\$ I'm hardly trying to replace something as intricate as Kafka or 0MQ. I'm putting forward a 16-line implementation of a publish-subscribe framework for local use. This code is far out of that league. That being said, I'll revise my question to add some more detail on what the code accomplishes and some possible points of critique. \$\endgroup\$
    – ndm13
    Aug 29, 2017 at 14:38

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.