EventManager


1   /*****************************************************************
2    *   Licensed to the Apache Software Foundation (ASF) under one
3    *  or more contributor license agreements.  See the NOTICE file
4    *  distributed with this work for additional information
5    *  regarding copyright ownership.  The ASF licenses this file
6    *  to you under the Apache License, Version 2.0 (the
7    *  "License"); you may not use this file except in compliance
8    *  with the License.  You may obtain a copy of the License at
9    *
10   *    http://www.apache.org/licenses/LICENSE-2.0
11   *
12   *  Unless required by applicable law or agreed to in writing,
13   *  software distributed under the License is distributed on an
14   *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15   *  KIND, either express or implied.  See the License for the
16   *  specific language governing permissions and limitations
17   *  under the License.
18   ****************************************************************/
19  
20  
21  package org.apache.cayenne.event;
22  
23  import java.util.ArrayList  ;
24  import java.util.Collections  ;
25  import java.util.EventObject  ;
26  import java.util.Iterator  ;
27  import java.util.LinkedList  ;
28  import java.util.List  ;
29  import java.util.Map  ;
30  import java.util.WeakHashMap  ;
31  
32  import org.apache.cayenne.CayenneRuntimeException;
33  import org.apache.cayenne.util.Invocation;
34  
35  /**
36   * This class acts as bridge between an Object that wants to inform others about its
37   * current state or a change thereof (Publisher) and a list of objects interested in the
38   * Subject (Listeners).
39   * 
40   * @author Dirk Olmes
41   * @author Holger Hoffstaette
42   * @author Andrus Adamchik
43   */
44  public class EventManager extends Object   {
45  
46      private static volatile EventManager defaultManager;
47  
48      public static final int DEFAULT_DISPATCH_THREAD_COUNT = 5;
49  
50      // keeps weak references to subjects
51      protected Map   subjects;
52      protected List   eventQueue;
53      protected boolean singleThread;
54  
55      /**
56       * Returns the shared EventManager. It is created on demand on the first call to this
57       * method. Cayenne internally doesn't use default EventManager. Instead Configuration
58       * EventManager is propagated to DataDomains and DataContexts.
59       * 
60       * @return EventManager the shared EventManager instance
61       */
62      public static EventManager getDefaultManager() {
63          if (defaultManager == null) {
64              synchronized (EventManager.class) {
65                  if (defaultManager == null) {
66                      defaultManager = new EventManager(2);
67                  }
68              }
69          }
70          return defaultManager;
71      }
72  
73      /**
74       * Creates a multithreaded EventManager using default thread count.
75       */
76      public EventManager() {
77          this(DEFAULT_DISPATCH_THREAD_COUNT);
78      }
79  
80      /**
81       * Creates an EventManager starting the specified number of threads for multithreaded
82       * dispatching. To create a single-threaded EventManager, use thread count of zero or
83       * less.
84       */
85      public EventManager(int dispatchThreadCount) {
86          this.subjects = Collections.synchronizedMap(new WeakHashMap  ());
87          this.eventQueue = Collections.synchronizedList(new LinkedList  ());
88          this.singleThread = dispatchThreadCount <= 0;
89  
90          // start dispatch threads
91          for (int i = 0; i < dispatchThreadCount; i++) {
92              new DispatchThread("EventDispatchThread-" + i).start();
93          }
94      }
95  
96      /**
97       * Returns true if this EventManager is single-threaded. If so it will throw an
98       * exception on any attempt to register an unblocking listener or dispatch a
99       * non-blocking event.
100      * 
101      * @since 1.2
102      */
103     public boolean isSingleThreaded() {
104         return singleThread;
105     }
106 
107     /**
108      * Returns a list of currently queued events. Queue is returned by copy. This method
109      * is useful for inspecting the state of the event queue at any particular moment, but
110      * doesn't allow callers to alter the queue state.
111      * 
112      * @since 1.1
113      */
114     public List   getEventQueue() {
115         synchronized (eventQueue) {
116             return new ArrayList  (eventQueue);
117         }
118     }
119 
120     /**
121      * Register an <code>EventListener</code> for events sent by any sender.
122      * 
123      * @throws RuntimeException if <code>methodName</code> is not found
124      * @see #addListener(Object, String, Class, EventSubject, Object)
125      */
126     public void addListener(
127             Object   listener,
128             String   methodName,
129             Class   eventParameterClass,
130             EventSubject subject) {
131         this.addListener(listener, methodName, eventParameterClass, subject, null, true);
132     }
133 
134     public void addNonBlockingListener(
135             Object   listener,
136             String   methodName,
137             Class   eventParameterClass,
138             EventSubject subject) {
139 
140         if (singleThread) {
141             throw new IllegalStateException  (
142                     "EventManager is configured to be single-threaded.");
143         }
144 
145         this.addListener(listener, methodName, eventParameterClass, subject, null, false);
146     }
147 
148     /**
149      * Register an <code>EventListener</code> for events sent by a specific sender.
150      * 
151      * @param listener the object to be notified about events
152      * @param methodName the name of the listener method to be invoked
153      * @param eventParameterClass the class of the single event argument passed to
154      *            <code>methodName</code>
155      * @param subject the event subject that the listener is interested in
156      * @param sender the object whose events the listener is interested in;
157      *            <code>null</code> means 'any sender'.
158      * @throws RuntimeException if <code>methodName</code> is not found
159      */
160     public void addListener(
161             Object   listener,
162             String   methodName,
163             Class   eventParameterClass,
164             EventSubject subject,
165             Object   sender) {
166         addListener(listener, methodName, eventParameterClass, subject, sender, true);
167     }
168 
169     public void addNonBlockingListener(
170             Object   listener,
171             String   methodName,
172             Class   eventParameterClass,
173             EventSubject subject,
174             Object   sender) {
175 
176         if (singleThread) {
177             throw new IllegalStateException  (
178                     "EventManager is configured to be single-threaded.");
179         }
180 
181         addListener(listener, methodName, eventParameterClass, subject, sender, false);
182     }
183 
184     protected void addListener(
185             Object   listener,
186             String   methodName,
187             Class   eventParameterClass,
188             EventSubject subject,
189             Object   sender,
190             boolean blocking) {
191 
192         if (listener == null) {
193             throw new IllegalArgumentException  ("Listener must not be null.");
194         }
195 
196         if (eventParameterClass == null) {
197             throw new IllegalArgumentException  ("Event class must not be null.");
198         }
199 
200         if (subject == null) {
201             throw new IllegalArgumentException  ("Subject must not be null.");
202         }
203 
204         try {
205             Invocation invocation = (blocking) ? new Invocation(
206                     listener,
207                     methodName,
208                     eventParameterClass) : new NonBlockingInvocation(
209                     listener,
210                     methodName,
211                     eventParameterClass);
212             dispatchQueueForSubject(subject, true).addInvocation(invocation, sender);
213         }
214         catch (NoSuchMethodException   nsm) {
215             throw new CayenneRuntimeException("Error adding listener, method name: "
216                     + methodName, nsm);
217         }
218     }
219 
220     /**
221      * Unregister the specified listener from all event subjects handled by this
222      * <code>EventManager</code> instance.
223      * 
224      * @param listener the object to be unregistered
225      * @return <code>true</code> if <code>listener</code> could be removed for any
226      *         existing subjects, else returns <code>false</code>.
227      */
228     public boolean removeListener(Object   listener) {
229         if (listener == null) {
230             return false;
231         }
232 
233         boolean didRemove = false;
234 
235         synchronized (subjects) {
236             if (!subjects.isEmpty()) {
237                 Iterator   subjectIter = subjects.keySet().iterator();
238                 while (subjectIter.hasNext()) {
239                     didRemove |= this.removeListener(listener, (EventSubject) subjectIter
240                             .next());
241                 }
242             }
243         }
244 
245         return didRemove;
246     }
247 
248     /**
249      * Removes all listeners for a given subject.
250      */
251     public boolean removeAllListeners(EventSubject subject) {
252         if (subject != null) {
253             synchronized (subjects) {
254                 return subjects.remove(subject) != null;
255             }
256         }
257 
258         return false;
259     }
260 
261     /**
262      * Unregister the specified listener for the events about the given subject.
263      * 
264      * @param listener the object to be unregistered
265      * @param subject the subject from which the listener is to be unregistered
266      * @return <code>true</code> if <code>listener</code> could be removed for the
267      *         given subject, else returns <code>false</code>.
268      */
269     public boolean removeListener(Object   listener, EventSubject subject) {
270         return this.removeListener(listener, subject, null);
271     }
272 
273     /**
274      * Unregister the specified listener for the events about the given subject and the
275      * given sender.
276      * 
277      * @param listener the object to be unregistered
278      * @param subject the subject from which the listener is to be unregistered
279      * @param sender the object whose events the listener was interested in;
280      *            <code>null</code> means 'any sender'.
281      * @return <code>true</code> if <code>listener</code> could be removed for the
282      *         given subject, else returns <code>false</code>.
283      */
284     public boolean removeListener(Object   listener, EventSubject subject, Object   sender) {
285         if (listener == null || subject == null) {
286             return false;
287         }
288 
289         DispatchQueue subjectQueue = dispatchQueueForSubject(subject, false);
290         if (subjectQueue == null) {
291             return false;
292         }
293 
294         return subjectQueue.removeInvocations(listener, sender);
295     }
296 
297     /**
298      * Sends an event to all registered objects about a particular subject. Event is sent
299      * synchronously, so the sender thread is blocked until all the listeners finish
300      * processing the event.
301      * 
302      * @param event the event to be posted to the observers
303      * @param subject the subject about which observers will be notified
304      * @throws IllegalArgumentException if event or subject are null
305      */
306     public void postEvent(EventObject   event, EventSubject subject) {
307         dispatchEvent(new Dispatch(event, subject));
308     }
309 
310     /**
311      * Sends an event to all registered objects about a particular subject. Event is
312      * queued by EventManager, releasing the sender thread, and is later dispatched in a
313      * separate thread.
314      * 
315      * @param event the event to be posted to the observers
316      * @param subject the subject about which observers will be notified
317      * @throws IllegalArgumentException if event or subject are null
318      * @since 1.1
319      */
320     public void postNonBlockingEvent(EventObject   event, EventSubject subject) {
321         if (singleThread) {
322             throw new IllegalStateException  (
323                     "EventManager is configured to be single-threaded.");
324         }
325 
326         // add dispatch to the queue and return
327         synchronized (eventQueue) {
328             eventQueue.add(new Dispatch(event, subject));
329             eventQueue.notifyAll();
330         }
331     }
332 
333     private void dispatchEvent(Dispatch dispatch) {
334         DispatchQueue dispatchQueue = dispatchQueueForSubject(dispatch.subject, false);
335         if (dispatchQueue != null) {
336             dispatchQueue.dispatchEvent(dispatch);
337         }
338     }
339 
340     // returns a subject's mapping from senders to registered listener invocations
341     private DispatchQueue dispatchQueueForSubject(EventSubject subject, boolean create) {
342         synchronized (subjects) {
343             DispatchQueue listenersStore = (DispatchQueue) subjects.get(subject);
344             if (create && listenersStore == null) {
345                 listenersStore = new DispatchQueue();
346                 subjects.put(subject, listenersStore);
347             }
348             return listenersStore;
349         }
350     }
351 
352     // represents a posted event
353     class Dispatch {
354 
355         EventObject  [] eventArgument;
356         EventSubject subject;
357 
358         Dispatch(EventObject   event, EventSubject subject) {
359             this(new EventObject  [] {
360                 event
361             }, subject);
362         }
363 
364         Dispatch(EventObject  [] eventArgument, EventSubject subject) {
365             this.eventArgument = eventArgument;
366             this.subject = subject;
367         }
368 
369         Object   getSender() {
370             return eventArgument[0].getSource();
371         }
372 
373         void fire() {
374             EventManager.this.dispatchEvent(Dispatch.this);
375         }
376 
377         boolean fire(Invocation invocation) {
378             if (invocation instanceof NonBlockingInvocation) {
379 
380                 // do minimal checks first...
381                 if (invocation.getTarget() == null) {
382                     return false;
383                 }
384 
385                 // inject single invocation dispatch into the queue
386                 synchronized (eventQueue) {
387                     eventQueue.add(new InvocationDispatch(
388                             eventArgument,
389                             subject,
390                             invocation));
391                     eventQueue.notifyAll();
392                 }
393 
394                 return true;
395             }
396             else {
397                 return invocation.fire(eventArgument);
398             }
399         }
400     }
401 
402     // represents a posted event that should be sent to a single known listener
403     class InvocationDispatch extends Dispatch {
404 
405         Invocation target;
406 
407         InvocationDispatch(EventObject  [] eventArgument, EventSubject subject,
408                 Invocation target) {
409             super(eventArgument, subject);
410             this.target = target;
411         }
412 
413         void fire() {
414             // there is no way to kill the invocation if it is bad...
415             // so don't check for status
416             target.fire(eventArgument);
417         }
418     }
419 
420     // subclass exists only to tag invocations that should be
421     // dispatched in a separate thread
422     final class NonBlockingInvocation extends Invocation {
423 
424         public NonBlockingInvocation(Object   target, String   methodName, Class   parameterType)
425                 throws NoSuchMethodException   {
426             super(target, methodName, parameterType);
427         }
428     }
429 
430     final class DispatchThread extends Thread   {
431 
432         DispatchThread(String   name) {
433             super(name);
434             setDaemon(true);
435         }
436 
437         public void run() {
438             while (true) {
439 
440                 // get event from the queue, if the queue
441                 // is empty, just wait
442                 Dispatch dispatch = null;
443 
444                 synchronized (EventManager.this.eventQueue) {
445                     if (EventManager.this.eventQueue.size() > 0) {
446                         dispatch = (Dispatch) EventManager.this.eventQueue.remove(0);
447                     }
448                     else {
449                         try {
450                             EventManager.this.eventQueue.wait();
451                         }
452                         catch (InterruptedException   e) {
453                             // ignore interrupts...
454                         }
455                     }
456                 }
457 
458                 // dispatch outside of synchronized block
459                 if (dispatch != null) {
460                     // this try/catch is needed to prevent DispatchThread
461                     // from dying on dispatch errors
462                     try {
463                         dispatch.fire();
464                     }
465                     catch (Throwable   th) {
466                         // ignoring exception
467                     }
468                 }
469             }
470         }
471     }
472 }
473
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags