listeneradaptor.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

     *
     * @param listener The listener to invoke. If null the resolution will take place, but obviously no listener will be invoked.
     * @param m The messenger being resolved.
     * @return true if the messenger was registered succesfully or the listener is null. If true it is garanteed that the listener
     * will be invoked unless null. If false, it is guaranteed that the listener will not be invoked.
     **/
    public boolean watchMessenger( MessengerEventListener listener, Messenger m ) {
        synchronized(this) {

            if (stopped) {
                return false;
            }

            if (listener == null) {
                // We're done, then. The invoker does not really care.
                return true;
            }

            // Init if needed.
            init();

            // First we must ensure that if the state changes we'll get to handle it.
            ListenerContainer allListeners = (ListenerContainer) inprogress.get(m.getIdentityReference());
            if (allListeners == null) {
                // Use ArrayList. The code is optimized for that.
                allListeners = new MessengerListenerContainer();
                inprogress.put(m.getIdentityReference(), allListeners);
            }
            allListeners.add(listener);
        }

        // When we do that, the selector get notified. Therefore we will always check the initial state automatically. If the
        // selectable is already done with, the listener will be called by the selector's handler.
        m.register(selector);

        return true;
    }

    /*
     * Any sort of listener type.
     */
    static abstract class ListenerContainer extends ArrayList {

        public ListenerContainer() {
            super(1);
        }
        protected abstract void giveUp(SimpleSelectable what, Throwable how);
        protected abstract void process(SimpleSelectable what);
    }

    /*
     * For messages
     */
    class MessageListenerContainer extends ListenerContainer {

        private void messageDone(Message m, OutgoingMessageEvent event) {

            // Note: synchronization is externally provided. When this method is invoked, this
            // object has already been removed from the map, so the list of listener cannot change.

            // Do not throw an iterator in the landfill for such a trivial case: optimize for an array list.
            int i = size();

            if (event == OutgoingMessageEvent.SUCCESS) {
                // Replace it with a msg-specific one.
                event = new OutgoingMessageEvent(m, null);

                while (i-->0) {
                    try {
                        ((OutgoingMessageEventListener) get(i)).messageSendSucceeded(event);
                    } catch (Throwable any) {
                        if (LOG.isEnabledFor(Level.WARN)) {
                            LOG.warn("Uncaught throwable in listener", any);
                        }
                    }
                }
                return;
            }

            if (event == OutgoingMessageEvent.OVERFLOW) {
                // Replace it with a msg-specific one.
                event = new OutgoingMessageEvent(m, null);
            }

            while (i-->0) {
                try {
                    ((OutgoingMessageEventListener) get(i)).messageSendFailed(event);
                } catch (Throwable any) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.warn("Uncaught throwable in listener", any);
                    }
                }
            }
        }

        protected void process(SimpleSelectable what) {

            Message m = (Message) what;

            OutgoingMessageEvent event = (OutgoingMessageEvent) m.getMessageProperty(Messenger.class);
            if (event == null) {
                return;
            }

            // Remove this container-selectable binding
            forgetSelectable(what);

            // Invoke app listeners
            messageDone(m, event);
        }

        protected void giveUp(SimpleSelectable what, Throwable how) {
            messageDone((Message) what, new OutgoingMessageEvent((Message) what, how));
        }
    }

    /**
     * For messengers
     **/
    class MessengerListenerContainer extends ListenerContainer {

        private void messengerDone(Messenger m) {

            // Note: synchronization is externally provided. When this method is invoked, this
            // object has already been removed from the map, so the list of listener cannot change.

            // Do not throw an iterator in the landfill for such a trivial case: optimize for an array list.
            int i = size();
            MessengerEvent event = new MessengerEvent(ListenerAdaptor.this, m, null);
            while (i-->0) {
                try {
                    ((MessengerEventListener) get(i)).messengerReady(event);
                } catch (Throwable any) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.warn("Uncaught throwable in listener", any);
                    }
                }
            }
        }

        protected void process(SimpleSelectable what) {

            Messenger m = (Messenger) what;

            if ((m.getState() & (Messenger.RESOLVED | Messenger.TERMINAL)) == 0) {
                return;
            }

            // Remove this container-selectable binding
            forgetSelectable(what);

            if ((m.getState() & Messenger.USABLE) == 0) {
                m = null;
            }

            // Invoke app listeners
            messengerDone(m);
        }

        protected void giveUp(SimpleSelectable what, Throwable how) {
            messengerDone(null);
        }
    }

    public void run() {
        try {
            while (!stopped) {
                List changed = selector.select();
                Iterator i = changed.iterator();
                while (i.hasNext()) {
                    SimpleSelectable m = (SimpleSelectable) i.next();
                    ListenerContainer listeners = null;
                    synchronized(this) {
                        listeners = (ListenerContainer) inprogress.get(m.getIdentityReference());
                    }
                    if (listeners == null) {
                        m.unregister(selector);
                        continue;
                    }
                    listeners.process(m);
                }
            }
        } catch(InterruptedException ie) {
        } catch (Throwable anyOther) {
            if (LOG.isEnabledFor(Level.FATAL)) {
                LOG.fatal("Uncaught throwable in background thread", anyOther);
                // There won't be any other thread. This thing is dead if that
                // happens. And it really shouldn't.
                synchronized(this) {
                    stopped = true;
                }
            }
        }


        // It's only us now. Stopped is true.
        IOException failed = new IOException("Endpoint interface terminated");
        Iterator i = inprogress.entrySet().iterator();
        while (i.hasNext()) {

            Map.Entry entry = (Map.Entry) i.next();
            SimpleSelectable m = (SimpleSelectable) ((SimpleSelectable.IdentityReference) entry.getKey()).getObject();
            ListenerContainer listeners = (ListenerContainer) entry.getValue();

            m.unregister(selector);

            if (listeners == null) {
                continue;
            }

            listeners.giveUp(m, failed);
        }
        inprogress.clear(); // Supposedly accelerates GC of entries.
    }
}