class TravelEvent extends Event {}      class FlightEvent extends TravelEvent {}      class LodgingEvent extends TravelEvent {}

      public class FlightDelayed        implements FlightEvent      {        private String message = "";

        public FlightDelayed()        {        }

        public FlightDelayed(String message)        {          this.message = message;        }

        public String getMessage()        {          return message;        }

        public void setMessage(String message)        {          this.message = message;        }      }

public interface Channel      {        void start();

        void stop();

        void publish(final Event event);

        void subscribe(final Receiver receiver,                       final Class eventClass);

        void unsubscribe(final Receiver receiver,                         final Class eventClass);      }

public class EventReceiver        implements Receiver      {        private static final transient Log log =          LogFactory.getLog(EventReceiver.class);

        private String id = "";

        public EventReceiver()        {        }

        public EventReceiver(String id)        {          this.id = id;        }

        public void setId(String id)        {          this.id = id;        }

        public String getId()        {          return id;        }

        public void receive(final Event event)        {          log.info("EventReceiver [" + id                   + "] received event [" + event.getMessage() + "]");        }      }

public class EventChannel        implements Channel      {        private static final String TOPIC_NAME =          "java:comp/env/EventTopic";

        private static final String MQ_URL = "tcp://localhost:61616";

        private HashMap subscribers = new HashMap();        private TopicConnectionFactory factory = null;        private Topic eventTopic = null;        private TopicConnection topicConn = null;        private TopicSession topicSess = null;        private TopicSubscriber topicSubscriber = null;        private TopicPublisher topicPublisher = null;        private EventConsumer eventConsumer = null;

        private void handleEvent(Event event)        {          final Set received = new HashSet();

          for (Class eventClass = event.getClass();               Event.class.isAssignableFrom(eventClass);               eventClass = eventClass.getSuperclass())          {            ArrayList receiverList = new ArrayList();            getReceiversForEvent(getEventLeafInterface(eventClass),                                 receiverList);            Receiver[] receivers = new Receiver[receiverList.size()];            receiverList.toArray(receivers);            for (int i = 0; i < receivers.length; i++)            {              invokeOnce(received, receivers[i], event);            }          }        }

        private void invokeOnce(Set received,                                Receiver receiver,                                Event event)        {          received.add(receiver);          receiver.receive(event);        }

        private Class getEventLeafInterface(Class cls)        {          Class retVal = null;

          if (Event.class.isAssignableFrom(cls))          {            retVal = cls;            if (cls.isInterface())            {              return retVal;            }          }

          Class[] interfaces = cls.getInterfaces();          if (interfaces != null)          {            for (int i = 0; i < interfaces.length; i++)            {              if (Event.class.isAssignableFrom(interfaces[i]))              {                retVal = interfaces[i];                break;              }              retVal = getEventLeafInterface(interfaces[i]);            }          }

          return retVal;        }

        public void start()        {          try          {            factory = new ActiveMQConnectionFactory(MQ_URL);            topicConn = factory.createTopicConnection();            topicSess =              topicConn.createTopicSession(false,                                           Session.AUTO_ACKNOWLEDGE);            eventTopic = topicSess.createTopic(TOPIC_NAME);            topicSubscriber = topicSess.createSubscriber(eventTopic);            topicPublisher = topicSess.createPublisher(eventTopic);

            eventConsumer = new EventConsumer(this);            Thread consumerThread = new Thread(eventConsumer);            consumerThread.setDaemon(false);            consumerThread.start();          }          catch (Exception e)          {            e.printStackTrace();          }        }

        public void stop()        {          // close topic connections, sessions, consumers, etc.        }

        public void publish(final Event event)        {          try          {            ObjectMessage eventMessage = topicSess.createObjectMessage();            eventMessage.setObject(event);

            topicPublisher.publish(eventMessage);          }          catch (Exception e)          {            e.printStackTrace();          }        }

        public void subscribe(final Receiver receiver,                              final Class eventClass)        {          ArrayList receiverList = null;

          Class leafCls = getEventLeafInterface(eventClass);

          if (subscribers.get(leafCls) == null)          {            receiverList = new ArrayList();            subscribers.put(leafCls, receiverList);          }          else          {            receiverList = (ArrayList) subscribers.get(leafCls);          }

          if (receiverList.indexOf(receiver) < 0)          {            receiverList.add(receiver);          }        }

        public void unsubscribe(final Receiver receiver,                                final Class eventClass)        {          Class leafCls = getEventLeafInterface(eventClass);          if (subscribers.get(leafCls) != null)          {            ArrayList receiverList = (ArrayList) subscribers.get(leafCls);            receiverList.remove(receiverList);          }        }      }

class EventConsumer       implements Runnable, ExceptionListener     {       private boolean running = false;       private boolean stopped = true;       private EventChannel eventChannel = null;

       private EventConsumer(EventChannel eventChannel)       {         this.eventChannel = eventChannel;       }

       public void run()       {         log.info("Event Consumer started");

         // Create a Topic Connection, Session, and a MessageConsumer for the Topic         // loop until stopped and distribute events to the event channel         // using the handleEvent method           eventChannel.handleEvent(event);

         stopped = true;

         log.info("Event Consumer stopped");       }

       public void shutdown()       {         running = false;

         while (stopped == false)         {           Thread.yield();         }       }     }

public class SenderServlet extends HttpServlet      {        private EventChannel eventChannel = null;        private EventReceiver allTravelEventReceiver = null;        private EventReceiver flightEventReceiver = null;        private EventReceiver lodgingEventReceiver = null;

        public void init()          throws ServletException        {          super.init();

          eventChannel = new EventChannel();          eventChannel.start();

          // create event receivers          allTravelEventReceiver =            new EventReceiver("allTravelEventReceiver");          flightEventReceiver =            new EventReceiver("flightEventReceiver");          lodgingEventReceiver =            new EventReceiver("lodgingEventReceiver");

          // subscribe to all Travel events          eventChannel.subscribe(allTravelEventReceiver,                                 TravelEvent.class);

          // subscribe to Flight events          eventChannel.subscribe(flightEventReceiver,                                 FlightEvent.class);

          // subscribe to Lodging events          eventChannel.subscribe(lodgingEventReceiver,                                 LodgingEvent.class);        }

        public void destroy()        {          super.destroy();

            // unsubscribe all event receivers and stop the event channel        }

        public void doGet(HttpServletRequest req, HttpServletResponse res)          throws IOException, ServletException        {          // respond with input form        }

        public void doPost(HttpServletRequest req, HttpServletResponse res)          throws IOException, ServletException        {          String txtMsg = req.getParameter("txtMsg");          if (txtMsg != null && txtMsg.length() > 0)          {            String flightDelayed = req.getParameter("FlightDelayed");            String rateIncreased = req.getParameter("RateIncreased");            String seatAvailable = req.getParameter("SeatAvailable");

            if (flightDelayed != null)            {              // send a Flight event              eventChannel.publish(new FlightDelayed(txtMsg));            }            else if (rateIncreased != null)            {              // send a Lodging event              eventChannel.publish(new RateIncreased(txtMsg));            }            else if (seatAvailable != null)            {              // send a Flight event              eventChannel.publish(new SeatAvailable(txtMsg));            }          }

          doGet(req, res);        }      }

00:12:33,921 INFO  [EventChannel] Starting EventChannel...00:12:33,937 INFO  [EventChannel] Creating topic connection...00:12:35,062 INFO  [EventChannel] EventChannel started00:12:35,062 INFO  [EventChannel] Event Consumer started00:12:35,093 INFO  [SenderServlet] AllTravelEventReceiver  [com.jeffhanson.eda.EventReceiver@f84033]00:12:35,093 INFO  [SenderServlet] FlightEventReceiver  [com.jeffhanson.eda.EventReceiver@3ee73b]00:12:35,093 INFO  [SenderServlet] LodgingEventReceiver  [com.jeffhanson.eda.EventReceiver@16127f4]

00:12:53,718 INFO  [SenderServlet] >>>>>00:12:53,718 INFO  [SenderServlet] >>>>>00:12:53,734 INFO  [EventChannel] Publishing event  [com.jeffhanson.eda.events.business.FlightDelayed@863854]00:12:53,859 INFO  [EventReceiver] EventReceiver [flightEventReceiver]  received event [Flight 2365 to Detroit will be delayed 15 minutes]00:12:53,859 INFO  [EventReceiver] EventReceiver [allTravelEventReceiver]  received event [Flight 2365 to Detroit will be delayed 15 minutes]