myserversocket.java

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

package connex.core.net;

import net.jxta.peergroup.PeerGroup;
import net.jxta.protocol.PipeAdvertisement;
import java.io.IOException;
import net.jxta.socket.JxtaServerSocket;
import java.net.Socket;
import java.net.SocketException;
import net.jxta.endpoint.Message;
import java.net.SocketTimeoutException;
import net.jxta.socket.JxtaSocket;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.document.StructuredDocument;
import net.jxta.endpoint.MessageElement;
import java.io.InputStream;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.AdvertisementFactory;
import org.apache.log4j.Logger;
import net.jxta.endpoint.Messenger;
import org.apache.log4j.Level;
import java.util.concurrent.TimeUnit;



/**
 * <p>Title: </p>
 * <p>Description: </p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: </p>
 * @author unbekannt
 * @version 1.0
 */

public class MyServerSocket extends JxtaServerSocket {

    private static final Logger LOG = Logger.getLogger(JxtaServerSocket.class.
            getName());
    public MyServerSocket(PeerGroup group, PipeAdvertisement pipeadv,
                          int backlog) throws
            IOException {
        super(group, pipeadv, backlog);
    }

    /**
  *  {@inheritDoc}
  *
  *  <p/>The default timeout is 60 seconds (60000ms).
  */
 public Socket accept() throws IOException {
     if (isClosed()) {
         throw new SocketException("Socket is closed");
     }
     if (!isBound()) {
         throw new SocketException("Socket is not bound yet");
     }
     try {
         if (LOG.isEnabledFor(Level.DEBUG)) {
             LOG.debug("Waiting for a connection");
         }
         while (true) {
             Message msg = (Message) queue.poll(timeout, TimeUnit.MILLISECONDS);

             if (msg == null) {
                 throw new SocketTimeoutException("Timeout reached");
             }
             JxtaSocket socket = processMessage(msg);

             // make sure we have a socket returning
             if (socket != null) {
                 if (LOG.isEnabledFor(Level.DEBUG)) {
                     LOG.debug("Waiting for a connection");
                 }
                 return socket;
             } else if (LOG.isEnabledFor(Level.DEBUG)) {
                 LOG.debug("No connection");
             }
         }
     } catch (InterruptedException ie) {
         throw new SocketException("interrupted");
     }
 }

 private JxtaSocket processMessage(Message msg) {

     PipeAdvertisement outputPipeAdv = null;
     PeerAdvertisement  peerAdv = null;
     StructuredDocument credDoc = null;

     if (LOG.isEnabledFor(Level.DEBUG)) {
         LOG.debug("Processing a connection message");
     }
     try {
         MessageElement el = msg.getMessageElement(nameSpace, credTag);

         if (el != null) {
             InputStream in = el.getStream();

             credDoc = (StructuredDocument)
                       StructuredDocumentFactory.newStructuredDocument(el.getMimeType(), in);
         }
         el = msg.getMessageElement(nameSpace, reqPipeTag);
         if (el != null) {
             InputStream in = el.getStream();

             outputPipeAdv = (PipeAdvertisement)
                             AdvertisementFactory.newAdvertisement(el.getMimeType(), in);
         }
         el = msg.getMessageElement(nameSpace, remPeerTag);
         if (el != null) {
             InputStream in = el.getStream();

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

         el = msg.getMessageElement(nameSpace, streamTag);
         boolean isStream = false;

         if (el != null) {
             isStream = (el.toString().equals("true"));
             if (LOG.isEnabledFor(Level.DEBUG)) {
                 LOG.debug("Connection request [isStream] :" + isStream);
             }
         }
         Messenger msgr = MySocket.lightweightOutputPipe1(group, outputPipeAdv, peerAdv);

         if (msgr != null) {
             PipeAdvertisement newpipe = newInputPipe(group, outputPipeAdv);
             JxtaSocket newsoc = new MySocket(group, msgr, newpipe, credDoc, isStream,peerAdv);

             sendResponseMessage(group, msgr, newpipe);
             return newsoc;
         }
     } catch (IOException e) {
         // deal with the error
         if (LOG.isEnabledFor(Level.DEBUG)) {
             LOG.debug("IOException occured", e);
         }
     }
     if (LOG.isEnabledFor(Level.DEBUG)) {
         LOG.debug("Connection processing did not result in a connection");
     }
     return null;
 }




}