messageprocessor.java

基于jxta的局域网P2P文件共享,可以实现局域网中的文件p2p共享,实现文件快速传输及交流

package connex.core.Presence;

import net.jxta.endpoint.*;
import connex.core.net.*;
import org.apache.log4j.Logger;
import org.apache.log4j.Level;
import net.jxta.document.MimeMediaType;
import java.util.*;
import net.jxta.document.XMLDocument;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.document.AdvertisementFactory;
import java.io.InputStream;
import java.io.IOException;
import java.io.FileInputStream;
import java.io.File;

import java.io.FileOutputStream;

public class MessageProcessor {
    private final static Logger LOG = org.apache.log4j.Logger.getLogger(
            MessageProcessor.class.getName());
    private WorkspaceConnection ses;

    /**
     * @directed
     */
    private MemberCollection members;

    /**
     * @associates connex.core.Presence.PresenceListener
     */
    /**
     * The vector containing the registered listeners which have to be notified.
     */

    private Vector presenceListenersVector = new java.util.Vector();
    //  private Member memb;
    private Hashtable<String,
            ServiceListener> services = new Hashtable<String, ServiceListener>();


    /**
     * Constructor
     * @param ses WorkspaceSession
     */
    protected MessageProcessor(WorkspaceConnection ses) {
        LOG.setLevel(Level.INFO);
        this.ses = ses;
        this.members = PresenceService.getInstance().getMemberCollection();
    }

    protected void process(Message msg) {
        String version;
        String membId = null;
        String backId = null;
        String type;
        String status = null;
        PeerAdvertisement peerAdv = null;
        ProfileAdv profileAdv = null;
        /* Message header*/

        MessageElement el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                                  PresenceProtocol.version);

        if (el != null) {
            version = el.toString();
        }

        el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                   PresenceProtocol.membIDTag);
        if (el != null) {
            membId = el.toString();
            if (membId.equals(PresenceService.getInstance().getmOwnPeerAdv().
                              getPeerID().toString())) {
                return;
            }
        }

        el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                   PresenceProtocol.backDoorTag);
        if (el != null) {
            backId = el.toString();
        }

        el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                   PresenceProtocol.typeTag);
        if (el != null) {
            type = el.toString();
            System.out.println();
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("Recived Message Type = " + type);
            }
            /*StatusUpdate Message*/

            if (type.equals(PresenceProtocol.statusUpdate)) {
                el = msg.getMessageElement(PresenceProtocol.nameSpace,PresenceProtocol.statusTag);
                if (el != null) {
                    status = el.toString();
                }

                // rigesterMember(membId, status, null, null, backId);
                this.processStatusUpdate(membId, status, backId);
                return;
            }

            /*pictureRequest Message*/
            if (type.equals(PresenceProtocol.pictureRequest)) {

                processPicReq(backId);
                return;
            }
            /*pictureResponse*/
            if (type.equals(PresenceProtocol.pictureResponse)) {

                processPicRes(membId, msg.getMessageElement(PresenceProtocol.nameSpace,
                        PresenceProtocol.pictureTag).
                              getBytes(false));
                return;
            }
            /*pictureUpdate Message*/
            if (type.equals(PresenceProtocol.pictureUpdate)) {

                processPicUpdate(membId, status, backId);
                return;
            }
            /*presenceRequest*/
            if (type.equals(PresenceProtocol.presenceRequest)) {
                el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                           PresenceProtocol.statusTag);
                if (el != null) {
                    status = el.toString();
                }

                profileAdv = this.getProfileAdvFromMessage(msg);
                peerAdv = this.getPeerAdvFromMessage(msg);
                processpresenceReq(membId, status, profileAdv, peerAdv, backId);

                return;
            }
            if (type.equals(PresenceProtocol.presenceResponse)) {
                el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                           PresenceProtocol.statusTag);
                if (el != null) {
                    status = el.toString();
                }

                profileAdv = this.getProfileAdvFromMessage(msg);
                peerAdv = this.getPeerAdvFromMessage(msg);
                processpresenceRes(membId, status, profileAdv, peerAdv, backId);

                return;
            }
            if (type.equals(PresenceProtocol.presenceUpdate)) {
                el = msg.getMessageElement(PresenceProtocol.nameSpace,
                                           PresenceProtocol.statusTag);
                if (el != null) {
                    status = el.toString();
                }

                profileAdv = this.getProfileAdvFromMessage(msg);
                peerAdv = this.getPeerAdvFromMessage(msg);
                processpresenceUpdate(membId, status, profileAdv, peerAdv,
                                     backId);

                return;
            }

            if (type.equals(PresenceProtocol.serviceRequest)) {

                processServiceReq(backId,
                                  msg.getMessageElement(PresenceProtocol.
                        nameSpace,
                        PresenceProtocol.serviceTag).toString());

                return;
            }
            if (type.equals(PresenceProtocol.serviceResponse)) {

                processServiceRes(membId,
                                  msg.getMessageElement(PresenceProtocol.
                        nameSpace,
                        PresenceProtocol.serviceTag).toString(),
                                  msg.getMessageElement(PresenceProtocol.
                        nameSpace,
                        PresenceProtocol.servicePipeTag).toString()
                        );

                return;
            }

        }

    }

    private ProfileAdv getProfileAdvFromMessage(Message msg) {
        ProfileAdv profileAdv = null;
        try {
            MessageElement el = msg.getMessageElement(PresenceProtocol.
                    nameSpace,
                    PresenceProtocol.profileTag);
            if (el != null) {

                InputStream in = el.getStream();
                profileAdv = (ProfileAdv)
                             AdvertisementFactory.newAdvertisement(
                                     el.
                                     getMimeType(), in);

            }

        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error(
                        "failed to process presenceRequest message",
                        e);
            }
        }

        return profileAdv;
    }

    private PeerAdvertisement getPeerAdvFromMessage(Message msg) {
        PeerAdvertisement peerAdv = null;
        try {
            MessageElement el = msg.getMessageElement(PresenceProtocol.
                    nameSpace,
                    PresenceProtocol.peerTag);
            if (el != null) {

                InputStream in = el.getStream();
                peerAdv = (PeerAdvertisement)
                          AdvertisementFactory.newAdvertisement(
                                  el.
                                  getMimeType(), in);

            }

        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error(
                        "failed to process presenceRequest message",
                        e);
            }
        }

        return peerAdv;
    }


    /**
     * propagete a PresenceRequestMessage within the Workspace.
     */
    protected void sendPresenceReq() {
        Message msg = PresenceProtocol.createPresenceMessage(PresenceProtocol.presenceRequest);

        try {

            ses.send(msg);
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info(" presenceReqMessage sent");
            }
        } catch (Exception ex) {

            ex.printStackTrace();
        }

    }

    /**
     * creats a PresenceRequestMessage and sending it direct to other Member.
     * @param backID String
     */
    protected void sendPresenceReq(String backID) {
        Message msg = PresenceProtocol.createPresenceMessage(PresenceProtocol.presenceRequest);

        try {

            ses.sendToBackDoor(backID, msg); // send with back door pipe
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("(backDoor) presenceReqMessage sent");
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

    }


    /**
     * creats a response to the RequestMessage and sending it direct to the requester.
     * @param backID String
     */
    protected void sendPresenceRes(String backID) {

        Message msg = PresenceProtocol.createPresenceMessage(PresenceProtocol.
                presenceResponse);

        try {

            ses.sendToBackDoor(backID, msg); // send with back door pipe
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info(" presenceRespondMessage sent");
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    protected void sendPresenceUpdate() {
        Message msg = PresenceProtocol.createPresenceMessage(PresenceProtocol.presenceUpdate);

        try {

            ses.send(msg);
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info(" presenceUpdateMessage sent");
            }
        } catch (Exception ex) {

            ex.printStackTrace();
        }

    }


    protected synchronized void sendStatusUpdate(String status) {
        Message msg = PresenceProtocol.createStatusUpdateMessage(status);

        ses.send(msg);
        if (LOG.isEnabledFor(Level.INFO)) {
            LOG.info(" StatusUpdateMessage sent");
        }

    }

    protected synchronized void sendPicReq(String backID) {
        Message msg = PresenceProtocol.createMessage(PresenceProtocol.pictureRequest);

        try {

            ses.sendToBackDoor(backID, msg); // send with back door pipe
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info(" pictureRequestMessage sent");
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

    }

    /**
     *
     * @param backID String
     */
    protected synchronized void sendPicRes(String backID) {
        Message msg = PresenceProtocol.createPictureResponseMessage();

        if (msg != null) {

            try {

                ses.sendToBackDoor(backID, msg); // send with back door pipe
                if (LOG.isEnabledFor(Level.INFO)) {
                    LOG.info(" pictureResponseMessage sent");
                }
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }