Mediation between Hibernate Entity Listeners and netty-socketio

Question

I am working on a small Java project, and here's the basic idea of what I'm trying to achieve: I have a layer over netty-socketio that accepts socket.io requests from the browser, and I use JPA 2.1/hibernate to persist the requested changes to DB. The twist is that I have a concept of stream requests as well, as in a user will request the current state of a collection and all future updates. To get the real-time updates from the DB, I'm using Entity Listeners. I'm looking for a solid way of connecting the entity listener methods to the handler on top of the socketio connection, i.e. the stream handler should be notified when the data it's interested in changed so it can send the update down the pipe. I tried to come up with a singleton moderator of sorts to which the entity listeners could post updates, and subscribed handlers could consume it, all based on a String topic, pretty much like a pubsub. The problem I hit was this: let's take for example the POJO User. When a new user is inserted, the UserEntityListener#PostInsert kicks in, and it forwards the user to the Notifier via a .publish call. The Notifier uses <?> for the data type, and it calls the interested parties via a Callable-like interface:

public interface Notifiable {
    public <T> void onEvent(T data);
}

So now the implementation of this is called in the proper handler, but it has the generic type and I have to cast it manually (the handler knows the type it should recieve). My question is, can I do this without explicit casts? Is there a good framework that makes all this low-level tinkering useless? I'd like a centralized solution to bridge the gap, otherwise all the boilerplate's going to kill me.

EDIT Added relevant source.

Notifier class:

class Subscriber {
    public String topic;
    public Notifiable notifiable;
    public Subscriber(String topic, Notifiable n) {
        this.topic = topic;
        this.notifiable = n;
    }
}

public class Notifier {
    private static Notifier instance = null;
    private List<Subscriber> subscribers = new ArrayList<Subscriber>();

    public Notifier() {};
    public void subscribe(String topic, Notifiable n) {
        if (!this.hasSubscriber(topic, n)) {
            this.subscribers.add(new Subscriber(topic, n));
        }
    }
    public <T> void publish(String topic, T data) {
        for (Subscriber s : this.subscribers) {
            if (s.topic.equals(topic)) {
                s.notifiable.onEvent(data);
            }
        }
    }
    public Boolean hasSubscriber (String topic, Notifiable n) {
        for (Subscriber s : this.subscribers) {
            if (s.topic.equals(topic) && s.notifiable == n) {
                return true;
            }
        }
        return false;
    }

    public static Notifier getInstance() {
        if (instance == null) {
            instance = new Notifier();
        }
        return instance;
    }
}

Entity Listener:

@PostPersist
public void PostInsert(User u) {
    Notifier.getInstance().publish("user/new", u);
}

Socketio Handler:

Notifier.getInstance().subscribe("user/new", (new Notifiable() {
    @Override
    public <T> void onEvent(T data) {
        User u = (User) data;
        logger.info("User name: " + u.getUsername());
    }
}));

Answer 1

If you want to avoid the explicit casting making the following changes:

One, Make your Notifiable interface generic:

public interface Notifiable<T> {
    public void onEvent(T data);
}

Two, make Subscriber class also generic:

public class Subscriber<T> {
    public String topic;
    public Notifiable<T> notifiable;
    public Subscriber(String topic, Notifiable<T> n) {
        ...
    }
 }

Three, adapt Notifier class

public class Notifier {
    private static Notifier instance = null;

    @SuppressWarnings("rawtypes")
    private List<Subscriber> subscribers = new ArrayList<Subscriber>();

    public Notifier() {};

    public <T> void  subscribe(String topic, Notifiable<T> n) {
        if (!this.hasSubscriber(topic, n)) {
            this.subscribers.add(new Subscriber<T>(topic, n));
        }
    }

    @SuppressWarnings("unchecked")
    public <T> void publish(String topic, T data) {
        for (Subscriber<T> s : this.subscribers) {
            if (s.topic.equals(topic)) {
                s.notifiable.onEvent(data);
            }
        }
    }

    @SuppressWarnings("unchecked")
    public <T> Boolean hasSubscriber (String topic, Notifiable<T> n) {
        for (Subscriber<T> s : this.subscribers) {
           /* XXX: Beware, is safe to compare two notifiable
            * instances by their memory addresses??
            */
           if (s.topic.equals(topic) && s.notifiable == n) {
                return true;
           }
        }
        return false;
    }

    public static Notifier getInstance() {
        if (instance == null) {
            instance = new Notifier();
        }
        return instance;
    }
}

Four, Socketio Handler:

Notifier.getInstance().subscribe("user/new", (new Notifiable<User>() {
    @Override
    public void onEvent(User data) {
        logger.info("User name: " + u.getUsername());
    }
}));