1 // Copyright (c) 2000 Just Objects B.V. <just@justobjects.nl>
2 // Distributable under LGPL license. See terms of license at gnu.org.
3 
4 package nl.justobjects.pushlet.core;
5 
6 import nl.justobjects.pushlet.util.Log;
7 
8 /**
9  * Routes Events to Subscribers.
10 *
11 * @author Just van den Broecke - Just Objects &copy;
12 * @version $Id: Dispatcher.java,v 1.8 2007/11/23 14:33:07 justb Exp $
13 */
14public class Dispatcher implements Protocol, ConfigDefs {
15    /**
16     * Singleton pattern:  single instance.
17     */
18    private static Dispatcher instance;
19
20    static {
21        try {
22            instance = (Dispatcher) Config.getClass(DISPATCHER_CLASS, "nl.justobjects.pushlet.core.Dispatcher").newInstance();
23            Log.info("Dispatcher created className=" + instance.getClass());
24        } catch (Throwable t) {
25            Log.fatal("Cannot instantiate Dispatcher from config", t);
26        }
27    }
28
29    /**
30     * Singleton pattern with factory method: protected constructor.
31     */
32    protected Dispatcher() {
33    }
34
35    /**
36     * Singleton pattern: get single instance.
37     */
38    public static Dispatcher getInstance() {
39        return instance;
40    }
41
42    /**
43     * Send event to all subscribers.
44     */
45    public synchronized void broadcast(Event event) {
46        // Get active sessions
47        Session[] sessions = getSessions();
48
49        // Send Event to all Subscribers
50        for (int i = 0; i < sessions.length; i++) {
51
52            // Snapshot array may not be filled entirely.
53            if (sessions[i] == null) {
54                break;
55            }
56            sessions[i].getSubscriber().onEvent((Event) event.clone());
57        }
58    }
59
60    /**
61     * Send event to subscribers matching Event subject.
62     */
63    public synchronized void multicast(Event event) {
64        // Get snapshot active sessions
65        Session[] sessions = getSessions();
66
67        // Send Event to all Subscribers whose Subject match Event
68        Event clonedEvent = null;
69        Subscription subscription = null;
70        Subscriber subscriber = null;
71        for (int i = 0; i < sessions.length; i++) {
72
73            // Snapshot array may not be filled entirely.
74            if (sessions[i] == null) {
75                break;
76            }
77
78            subscriber = sessions[i].getSubscriber();
79
80            // Send only if the subscriber's criteria
81            // match the event.
82            if ((subscription = subscriber.match(event)) != null) {
83                // Personalize event
84                clonedEvent = (Event) event.clone();
85
86                // Set subscription id and optional label
87                clonedEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
88                if (subscription.getLabel() != null) {
89                    event.setField(P_SUBSCRIPTION_LABEL, subscription.getLabel());
90                }
91
92                subscriber.onEvent(clonedEvent);
93            }
94        }
95
96    }
97
98    /**
99     * Send event to specific subscriber.
00     */
01    public synchronized void unicast(Event event, String aSessionId) {
02        // Get subscriber to send event to
03        Session session = SessionManager.getInstance().getSession(aSessionId);
04        if (session == null) {
05            Log.warn("unicast: session with id=" + aSessionId + " does not exist");
06            return;
07        }
08
09        // Send Event to subscriber.
10        session.getSubscriber().onEvent((Event) event.clone());
11    }
12
13    /**
14     * Start Dispatcher.
15     */
16    public void start() {
17        Log.info("Dispatcher started");
18    }
19
20    /**
21     * Stop Dispatcher.
22     */
23    public void stop() {
24        // Send abort control event to all subscribers.
25        Log.info("Dispatcher stopped: broadcast abort to all subscribers");
26        broadcast(new Event(E_ABORT));
27    }
28
29    private Session[] getSessions() {
30        return SessionManager.getInstance().getSnapshot();
31    }
32}
33
34/*
35 * $Log: Dispatcher.java,v $
36 * Revision 1.8  2007/11/23 14:33:07  justb
37 * core classes now configurable through factory
38 *
39 * Revision 1.7  2005/02/28 12:45:59  justb
40 * introduced Command class
41 *
42 * Revision 1.6  2005/02/28 09:14:55  justb
43 * sessmgr/dispatcher factory/singleton support
44 *
45 * Revision 1.5  2005/02/21 16:59:06  justb
46 * SessionManager and session lease introduced
47 *
48 * Revision 1.4  2005/02/21 11:50:46  justb
49 * ohase1 of refactoring Subscriber into Session/Controller/Subscriber
50 *
51 * Revision 1.3  2005/02/18 12:36:47  justb
52 * changes for renaming and configurability
53 *
54 * Revision 1.2  2005/02/18 10:07:23  justb
55 * many renamings of classes (make names compact)
56 *
57 * Revision 1.1  2005/02/18 09:54:15  justb
58 * refactor: rename Publisher Dispatcher and single Subscriber class
59 *
60 * Revision 1.14  2005/02/16 14:39:34  justb
61 * fixed leave handling and added "poll" mode
62 *
63 * Revision 1.13  2004/10/24 20:50:35  justb
64 * refine subscription with label and sending sid and label on events
65 *
66 * Revision 1.12  2004/10/24 12:58:18  justb
67 * revised client and test classes for new protocol
68 *
69 * Revision 1.11  2004/09/26 21:39:43  justb
70 * allow multiple subscriptions and out-of-band requests
71 *
72 * Revision 1.10  2004/09/20 22:01:38  justb
73 * more changes for new protocol
74 *
75 * Revision 1.9  2004/09/03 22:35:37  justb
76 * Almost complete rewrite, just checking in now
77 *
78 * Revision 1.8  2004/08/13 23:36:05  justb
79 * rewrite of Pullet into Pushlet "pull" mode
80 *
81 * Revision 1.7  2004/08/12 13:18:54  justb
82 * cosmetic changes
83 *
84 * Revision 1.6  2004/03/10 15:45:55  justb
85 * many cosmetic changes
86 *
87 * Revision 1.5  2004/03/10 13:59:28  justb
88 * rewrite using Collection classes and finer synchronization
89 *
90 * Revision 1.4  2003/08/15 08:37:40  justb
91 * fix/add Copyright+LGPL file headers and footers
92 *
93 * Revision 1.3  2003/08/12 08:54:40  justb
94 * added getSubscriberCount() and use Log
95 *
96 * Revision 1.2  2003/05/18 16:15:08  justb
97 * support for XML encoded Events
98 *
99 * Revision 1.1.1.1  2002/09/24 21:02:31  justb
00 * import to sourceforge
01 *
02 * Revision 1.1.1.1  2002/09/20 22:48:18  justb
03 * import to SF
04 *
05 * Revision 1.1.1.1  2002/09/20 14:19:04  justb
06 * first import into SF
07 *
08 * Revision 1.3  2002/04/15 20:42:41  just
09 * reformatting and renaming GuardedQueue to EventQueue
10 *
11 * Revision 1.2  2000/08/21 20:48:29  just
12 * added CVS log and id tags plus copyrights
13 *
14 *
15 */
16