A generic observer pattern in Java

Question

The java.util.Observer and java.util.Observable are ugly. They require the sorts of casts that make type-safety fans uncomfortable, and you can't define a class to be an Observer of multiple things without ugly casts. In fact, in "How do I know the generic object that the Observer class sends in Java?", an answerer says that only one type of data should be used in each observer / observable.

I'm trying to make a generic version of the observer pattern in Java to get round both these problems. It's not unlike the one in the previously mentioned post, but that question was not obviously resolved (the last comment is an unanswered question from the OP).

Answer 1

Observer.java

package util;

public interface Observer<ObservedType> {
    public void update(Observable<ObservedType> object, ObservedType data);
}

Observable.java

package util;

import java.util.LinkedList;
import java.util.List;

public class Observable<ObservedType> {

    private List<Observer<ObservedType>> _observers = 
      new LinkedList<Observer<ObservedType>>();

    public void addObserver(Observer<ObservedType> obs) {
        if (obs == null) {
            throw new IllegalArgumentException("Tried
                      to add a null observer");
        }
        if (_observers.contains(obs)) {
            return;
        }
        _observers.add(obs);
    }

    public void notifyObservers(ObservedType data) {
        for (Observer<ObservedType> obs : _observers) {
            obs.update(this, data);
        }
    }
}

Hopefully this will be useful to someone.

Answer 2

I prefer using an annotation so a listener can listen to different types of events.

public class BrokerTestMain {
    public static void main(String... args) {
        Broker broker = new Broker();
        broker.add(new Component());

        broker.publish("Hello");
        broker.publish(new Date());
        broker.publish(3.1415);
    }
}

class Component {
    @Subscription
    public void onString(String s) {
        System.out.println("String - " + s);
    }

    @Subscription
    public void onDate(Date d) {
        System.out.println("Date - " + d);
    }

    @Subscription
    public void onDouble(Double d) {
        System.out.println("Double - " + d);
    }
}

prints

String - Hello
Date - Tue Nov 13 15:01:09 GMT 2012
Double - 3.1415

---

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Subscription {
}

public class Broker {
    private final Map<Class, List<SubscriberInfo>> map = new LinkedHashMap<Class, List<SubscriberInfo>>();

    public void add(Object o) {
        for (Method method : o.getClass().getMethods()) {
            Class<?>[] parameterTypes = method.getParameterTypes();
            if (method.getAnnotation(Subscription.class) == null || parameterTypes.length != 1) continue;
            Class subscribeTo = parameterTypes[0];
            List<SubscriberInfo> subscriberInfos = map.get(subscribeTo);
            if (subscriberInfos == null)
                map.put(subscribeTo, subscriberInfos = new ArrayList<SubscriberInfo>());
            subscriberInfos.add(new SubscriberInfo(method, o));
        }
    }

    public void remove(Object o) {
        for (List<SubscriberInfo> subscriberInfos : map.values()) {
            for (int i = subscriberInfos.size() - 1; i >= 0; i--)
                if (subscriberInfos.get(i).object == o)
                    subscriberInfos.remove(i);
        }
    }

    public int publish(Object o) {
        List<SubscriberInfo> subscriberInfos = map.get(o.getClass());
        if (subscriberInfos == null) return 0;
        int count = 0;
        for (SubscriberInfo subscriberInfo : subscriberInfos) {
            subscriberInfo.invoke(o);
            count++;
        }
        return count;
    }

    static class SubscriberInfo {
        final Method method;
        final Object object;

        SubscriberInfo(Method method, Object object) {
            this.method = method;
            this.object = object;
        }

        void invoke(Object o) {
            try {
                method.invoke(object, o);
            } catch (Exception e) {
                throw new AssertionError(e);
            }
        }
    }
}

Answer 3

A modern update: ReactiveX is a very nice API for asynchronous programming based on the Observer pattern, and it's fully generic. If you're using Observer/Observable to "stream" data or events from one place in your code to another, you should definitely look into it.

It's based on functional programming, so it looks very sleek with Java 8's lambda syntax:

Observable.from(Arrays.asList(1, 2, 3, 4, 5))
        .reduce((x, y) -> x + y)
        .map((v) -> "DecoratedValue: " + v)
        .subscribe(System.out::println);