listeneradaptor.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

            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.
            MessengerListenerContainer allListeners = (MessengerListenerContainer) inprogress.get(messenger.getIdentityReference());

            if (allListeners == null) {
                // Use ArrayList. The code is optimized for that.
                allListeners = new MessengerListenerContainer();
                inprogress.put(messenger.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.
        messenger.register(selector);

        return true;
    }

    /*
     * Any sort of listener type.
     */
    static abstract class ListenerContainer<S extends SimpleSelectable, L extends java.util.EventListener> extends ArrayList<L> {

        public ListenerContainer() {
            super(1);
        }

        protected abstract void giveUp(S what, Throwable how);

        protected abstract void process(S what);
    }


    /**
     * For messages
     */
    @SuppressWarnings("serial")
    class MessageListenerContainer extends ListenerContainer<Message, OutgoingMessageEventListener> {

        private void messageDone(Message message, 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.

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

                for (OutgoingMessageEventListener eachListener : this) {
                    try {
                        eachListener.messageSendSucceeded(event);
                    } catch (Throwable any) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "Uncaught throwable from listener", any);
                        }
                    }
                }
            } else {
                if (event == OutgoingMessageEvent.OVERFLOW) {
                    // Replace it with a msg-specific one.
                    event = new OutgoingMessageEvent(message, null);
                }

                for (OutgoingMessageEventListener eachListener : this) {
                    try {
                        eachListener.messageSendFailed(event);
                    } catch (Throwable any) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "Uncaught throwable in listener", any);
                        }
                    }
                }
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        protected void process(Message message) {
            OutgoingMessageEvent event = (OutgoingMessageEvent) message.getMessageProperty(Messenger.class);

            if (event == null) {
                return;
            }

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

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

        /**
         * {@inheritDoc}
         */
        @Override
        protected void giveUp(Message m, Throwable how) {
            messageDone(m, new OutgoingMessageEvent(m, how));
        }
    }

    /**
     * For messengers
     */
    @SuppressWarnings("serial")
    class MessengerListenerContainer extends ListenerContainer<Messenger, MessengerEventListener> {

        private void messengerDone(Messenger messenger) {

            // 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.

            MessengerEvent event = new MessengerEvent(ListenerAdaptor.this, messenger, null);

            for (MessengerEventListener eachListener : this) {
                try {
                    eachListener.messengerReady(event);
                } catch (Throwable any) {
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.log(Level.WARNING, "Uncaught throwable in listener", any);
                    }
                }
            }
        }

        /**
         * {@inheritDoc}
         */
        @Override
        protected void process(Messenger messenger) {
            if ((messenger.getState() & (Messenger.RESOLVED | Messenger.TERMINAL)) == 0) {
                return;
            }

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

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

            // Invoke app listeners
            messengerDone(messenger);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        protected void giveUp(Messenger m, Throwable how) {
            messengerDone(null);
        }
    }

    /**
     * {@inheritDoc}
     */
    public void run() {
        try {
            while (!shutdown) {
                try {
                    Collection<SimpleSelectable> changed = selector.select();
                    for (SimpleSelectable simpleSelectable : changed) {
                        ListenerContainer listeners;
                        synchronized (this) {
                            listeners = inprogress.get(simpleSelectable.getIdentityReference());
                        }
                        if (listeners == null) {
                            simpleSelectable.unregister(selector);
                            continue;
                        }
                        if (executor == null) {
                            listeners.process(simpleSelectable);
                        } else {
                            executor.execute(new ListenerProcessor(listeners, simpleSelectable));
                        }
                    }
                } catch (InterruptedException ie) {
                    Thread.interrupted();
                }
            }
        } catch (Throwable anyOther) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "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) {
                shutdown = true;
            }
        } finally {
            try {
                // It's only us now. Stopped is true.
                IOException failed = new IOException("Endpoint interface terminated");
                for (Map.Entry<IdentityReference, ListenerContainer> entry : inprogress.entrySet()) {
                    SimpleSelectable simpleSelectable = entry.getKey().getObject();
                    ListenerContainer listeners = entry.getValue();
                    simpleSelectable.unregister(selector);

                    if (listeners != null) {
                        listeners.giveUp(simpleSelectable, failed);
                    }
                }
                inprogress.clear();
            } catch (Throwable anyOther) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, "Uncaught Throwable while shutting down background thread", anyOther);
                }
            }
            bgThread = null;
        }
    }

    /**
     * A small class for processing individual messages.
     */
    private class ListenerProcessor implements Runnable {

        private SimpleSelectable simpleSelectable;
        private ListenerContainer listeners;
        ListenerProcessor(ListenerContainer listeners, SimpleSelectable simpleSelectable) {
            this.listeners = listeners;
            this.simpleSelectable = simpleSelectable;
        }

        public void run() {
            listeners.process(simpleSelectable);
        }
    }
}