remotecommandinvoker.java

用jxse开发的一个p2p通讯软件 有聊天 文件共享 视频3大功能

/*
*  Copyright (c) 2001 Sun Microsystems, Inc.  All rights
*  reserved.
*
*  Redistribution and use in source and binary forms, with or without
*  modification, are permitted provided that the following conditions
*  are met:
*
*  1. Redistributions of source code must retain the above copyright
*  notice, this list of conditions and the following disclaimer.
*
*  2. Redistributions in binary form must reproduce the above copyright
*  notice, this list of conditions and the following discalimer in
*  the documentation and/or other materials provided with the
*  distribution.
*
*  3. The end-user documentation included with the redistribution,
*  if any, must include the following acknowledgment:
*  "This product includes software developed by the
*  Sun Microsystems, Inc. for Project JXTA."
*  Alternately, this acknowledgment may appear in the software itself,
*  if and wherever such third-party acknowledgments normally appear.
*
*  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA"
*  must not be used to endorse or promote products derived from this
*  software without prior written permission. For written
*  permission, please contact Project JXTA at http://www.jxta.org.
*
*  5. Products derived from this software may not be called "JXTA",
*  nor may "JXTA" appear in their name, without prior written
*  permission of Sun.
*
*  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
*  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
*  OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
*  DISCLAIMED.  IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
*  ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
*  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
*  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
*  USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
*  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
*  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
*  OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
*  SUCH DAMAGE.
*  ====================================================================
*
*  This software consists of voluntary contributions made by many
*  individuals on behalf of Project JXTA.  For more
*  information on Project JXTA, please see
*  <http://www.jxta.org/>.
*
*  This license is based on the BSD license adopted by the Apache Foundation.
*
*  $Id: RemoteCommandInvoker.java,v 1.5 2007/05/28 22:00:52 nano Exp $
*/

package net.jxta.myjxta.dialog.util;

import net.jxta.endpoint.MessageElement;
import net.jxta.logging.Logging;
import net.jxta.myjxta.MyJXTA;
import net.jxta.myjxta.dialog.*;
import net.jxta.myjxta.util.Group;
import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PipeAdvertisement;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author james todd [gonzo at jxta dot org]
 * @version $Id: RemoteCommandInvoker.java,v 1.5 2007/05/28 22:00:52 nano Exp $
 */

public class RemoteCommandInvoker {

    private final Logger LOG = Logger.getLogger(RemoteCommandInvoker.class.getName());

    private Group group = null;
    private PipeAdvertisement pipeAdvertisement = null;
    private Command command = null;
    private MyJXTA myjxta = null;
    private DialogMessage response = null;
    private IOException m_pipeExeption = null;


    public RemoteCommandInvoker(Group g, PipeAdvertisement pa, Command c,
                                MyJXTA myjxta) {
        this.group = g;
        this.pipeAdvertisement = pa;
        this.command = c;
        this.myjxta = myjxta;

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("RemoteCommandInvoker instantiated");
        }
    }

    public DialogMessage getResponse() {
        DialogMessage r = this.response;

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("response: " + r);
        }

        return r;
    }

    /**
     * invokes the command with the specified timeout on the remote peer
     *
     * @param timeoutInMillis
     */
    public void invoke(long timeoutInMillis) {
        invoke(250, timeoutInMillis);
    }

    private void invoke(long interval, long max) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("fetch[interval, max]: " + interval + " " + max);
        }

        sendCommand(interval, max, null);
    }

    /**
     * sends the command with the additional message elements and the configured
     * timeout to the remote peer.
     *
     * @param timeoutInMillis
     * @param messageElements
     */
    public void invoke(long timeoutInMillis, Map<String, MessageElement> messageElements) {
        sendCommand(250, timeoutInMillis, messageElements);
    }

    private void sendCommand(long interval, long max, Map<String, MessageElement> messages) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("fetch[interval, max, messages]: " + interval + " " + max);
        }

        PeerGroup pg = this.group.getPeerGroup();
        OneToOneCommandDialog d = DialogManager.getCommandDialog(OneToOneCommandDialog.class,
                this.group, this.pipeAdvertisement, this.myjxta);
        d.activate();
        if (d.isConnected()) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("add listener");
            }

            DialogListener myDialogListener = new DialogListener() {

                public void receive(DialogMessage msg) {
                    response = msg;
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("received msg: " + msg);
                    }
                }
            };
            d.addListener(myDialogListener);

            DialogMessage request = new DialogMessage(pg.getPeerName(), null,
                    pg.getPeerGroupID().toString(), pg.getPeerGroupName());

            request.setCommand(this.command.getRequest());

            if (messages != null) {
                Set<String> keySet = messages.keySet();
                for (String aKeySet : keySet) {
                    request.addMessageElement(aKeySet, messages.get(aKeySet));
                }
            }

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("request dialog: " + request);
            }

            d.dispatch(request);
            //nano: doesnt make sense to wait for an answer if we dont have a pipe to send the request does it?
            if (!d.isConnected())
                return;


            long m = System.currentTimeMillis() + max;

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("waiting: " + max);
            }

            while (this.response == null &&
                    m > System.currentTimeMillis()) {
                try {
                    Thread.sleep(interval);
                } catch (InterruptedException ie) {
                    m = 0;

                    if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                        LOG.log(Level.SEVERE, "sleep interrupt: ", ie);
                    }
                }
            }
            d.removeListener(myDialogListener);
            m_pipeExeption = null;
        } else {
            m_pipeExeption = d.getPipeCreationException();
        }

    }


    public IOException getPipeExeption() {
        return m_pipeExeption;
    }

}