memberconnection.java

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

package connex.core.net;

import net.jxta.socket.JxtaSocket;
import java.net.*;
import java.io.*;

import net.jxta.endpoint.Message;
import net.jxta.socket.JxtaSocketAddress;
import net.jxta.id.ID;
import net.jxta.pipe.PipeID;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.peer.PeerID;
import org.apache.log4j.Logger;
import connex.core.WS.Workspace;


public class MemberConnection extends Connection implements Runnable {

    private JxtaSocket soc;
    private InputStream in;
    private OutputStream out;
    private byte[] buffer;
    private int buffSize = 46 * 1024;
    private PipeID id = null;
    private PeerID pId = null;
    private Thread t=null;
    static final Logger LOG = Logger.getLogger(MemberConnection.class);
    public MemberConnection(Workspace ws, String peerID, String pipeID,
                         ConnectionClient listener) {
        super(ws.getPeerGroup(), listener);

        try {
            this.id = (PipeID) ID.create(new URI(pipeID));
            this.pId = (PeerID) ID.create(new URI(peerID));
        } catch (URISyntaxException ex) {
        }
    }

    public MemberConnection(MySocket soc) {
        this.soc = soc;

    }

    public void setClient(ConnectionClient listener) {
        this.listener = listener;

    }


    public boolean connect() {
        if(soc!=null){
             init();
          return soc.isConnected();
        }
        PipeAdvertisement pAdv = PipeUtils.createPipeAdv("ConneXPipe", "ConneXServicePipe");
        pAdv.setPipeID(id);
        try {
            soc = new JxtaSocket(new JxtaSocketAddress(pg, pAdv,pId), 30000);
            init();
            return soc.isConnected();
        } catch (IOException ex1) {
            System.err.println(ex1.getMessage());
        }
        return false;
    }

    private void init() {
        if (soc != null) {
            try {

                soc.setSoTimeout(0);

                soc.setOutputStreamBufferSize(46 * 1024);

                soc.setReceiveBufferSize(46 * 1024);

                in = soc.getInputStream();
                out = soc.getOutputStream();
                    System.out.println("in out initialized ");
                System.out.println("Client Connected");
                if(t==null){
                    t = new Thread(this);
                    t.start();
                }
            } catch (IOException ex1) {
                System.err.println(ex1.getMessage());
            }

        }

    }

    public String getRemotPeerID() {

        return ((MySocket) soc).getRemotePeerID();
    }

    public String getRemotePeerName() {

        return ((MySocket) soc).getRemotePeerName();
    }

    public synchronized void send(Message msg) {

        try {

            byte[] data = ObjectUtil.objectToBytes(msg);

            out.write(data);

            out.flush();

             sent++;
        } catch (Exception io) {
            System.out.println(io.getMessage() + " can't send the Message");
        }

    }


    public void run() {

        buffer = new byte[buffSize];

        while (soc.isBound()) {
            try {
                in.read(buffer);
                Message msg = null;
                try {
                    msg = (Message) ObjectUtil.bytesToObject(buffer);
                    listener.reciveMessage(msg);
                    recived++;
                } catch (ClassNotFoundException ex) {}

            } catch (IOException io) {
                System.err.println("linein " + io.getMessage());

            }

        }
        if (listener != null) {
            System.out.println("ClosefromRemote:");
            listener.closedFromRemote();
            disConnect();
        }
    }

    public ConnectionClient getOwner() {
        return listener;
    }

    public long getSentMessageCount() {
        return sent;
    }

    public long getRecivedMessageCount() {
        return recived;
    }

    public boolean disConnect() {

        if (soc.isConnected()) {

            try {
                listener = null;
                soc.close();

            } catch (IOException ex2) {
                System.err.println("close" + ex2.getMessage());
                ex2.printStackTrace();

            }

        }
        System.out.println("is closed: " + soc.isClosed());

        return soc.isClosed();
    }


}