server.java

javaP2P技术内幕课程111213141516源代码

/*
 * Created by IntelliJ IDEA.
 * User: fsommers
 * Date: Apr 7, 2002
 * Time: 6:07:09 PM
 * To change template for new class use
 * Code Style | Class Templates options (Tools | IDE Options).
 */
package primecruncher;

import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.discovery.DiscoveryService;
import net.jxta.pipe.*;
import net.jxta.exception.PeerGroupException;
import net.jxta.protocol.ModuleClassAdvertisement;
import net.jxta.protocol.ModuleSpecAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.document.*;
import net.jxta.platform.ModuleClassID;
import net.jxta.platform.ModuleSpecID;
import net.jxta.id.IDFactory;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;

import java.io.*;

public class Server {

    private static PeerGroup group;
    private static DiscoveryService discoSvc;
    private static PipeService pipeSvc;

    private InputPipe inputPipe;

    private static final String PIPE_ADV_FILE = "primeserver_pipe.adv";


    public static void main(String[] argv) {
        Server server = new Server();
        server.startJxta();
        server.doAdvertise();
        server.startService();

    }

    public Server() {

    }

    private void doAdvertise() {

        ModuleClassAdvertisement classAd =
                (ModuleClassAdvertisement) AdvertisementFactory.newAdvertisement(
                        ModuleClassAdvertisement.getAdvertisementType());

        ModuleClassID classID = IDFactory.newModuleClassID();
        classAd.setModuleClassID(classID);
        classAd.setName(ServiceConstants.CLASS_NAME);
        classAd.setDescription("A prime number crunching service.");

        try {
            discoSvc.publish(classAd, DiscoveryService.ADV);
            discoSvc.remotePublish(classAd, DiscoveryService.ADV);
            System.out.println("Published module class adv.");
        } catch (IOException e) {
            System.out.println("Trouble publishing module class adv: " + e.getMessage());
        }


        ModuleSpecAdvertisement specAd =
                (ModuleSpecAdvertisement) AdvertisementFactory.newAdvertisement(
                        ModuleSpecAdvertisement.getAdvertisementType());

        ModuleSpecID specID = IDFactory.newModuleSpecID(classID);
        specAd.setModuleSpecID(specID);
        specAd.setName(ServiceConstants.SPEC_NAME);
        specAd.setDescription("Specification for a prime number crunching service");
        specAd.setCreator("Sams Publishing");
        specAd.setSpecURI("http://www.samspulishing.com/p2p/primecruncher");
        specAd.setVersion("Version 1.0");


        PipeAdvertisement pipeAd = null;

        try {
            FileInputStream is = new FileInputStream(PIPE_ADV_FILE);
            pipeAd = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                    new MimeMediaType("text/xml"), is);
            is.close();
            System.out.println("Read pipe advert from disk.");
        } catch (IOException e) {
            pipeAd = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                    PipeAdvertisement.getAdvertisementType());
            PipeID pid = IDFactory.newPipeID(group.getPeerGroupID());
            pipeAd.setPipeID(pid);
            //save pipeAd in file
            Document pipeAdDoc = pipeAd.getDocument(new MimeMediaType("text/xml"));
            try {
                FileOutputStream os = new FileOutputStream(PIPE_ADV_FILE);
                pipeAdDoc.sendToStream(os);
                os.flush();
                os.close();
                System.out.println("Wrote pipe advertisement to disk.");
            } catch (IOException ex) {
                System.out.println("Can't save pipe advertisement to file " + PIPE_ADV_FILE);
                System.exit(-1);
            }
        }

        specAd.setPipeAdvertisement(pipeAd);

        try {
            StructuredTextDocument doc = (StructuredTextDocument)
                    specAd.getDocument(new MimeMediaType("text/plain"));
            StringWriter out = new StringWriter();
            doc.sendToWriter(out);
            System.out.println(out);
            out.close();
        } catch (IOException e) {
            e.printStackTrace();
        }


        try {
            discoSvc.publish(specAd, DiscoveryService.ADV);
            discoSvc.remotePublish(specAd, DiscoveryService.ADV);
            System.out.println("Published module spec adv");
        } catch (IOException e) {
            System.out.println("Trouble publishing module spec adv: " + e.getMessage());
        }


        //create an input pipe based on the advertisement
        try {
            inputPipe = pipeSvc.createInputPipe(pipeAd);
            System.out.println("Created input pipe");
        } catch (IOException e) {
            System.out.println("Can't create input pipe. " + e.getMessage());
        }

    }

    private void startService() {

        while (true) {

            Message msg = null;

            try {
                msg = inputPipe.waitForMessage();
            } catch (InterruptedException ex) {
                inputPipe.close();
                return;
            }

            MessageElement hiElem = msg.getElement(ServiceConstants.HIGH_INT);
            String highInt = new String(hiElem.getBytesOffset());
            MessageElement loElem = msg.getElement(ServiceConstants.LOW_INT);
            String lowInt = new String(loElem.getBytesOffset());
            MessageElement idEl = msg.getElement(ServiceConstants.JOBID);
            String jobID = new String(idEl.getBytesOffset());


            if (highInt != null || lowInt != null) {
                final Message res = processInput(jobID, highInt, lowInt);

                if (res != null) {
                    try {
                        //open a pipe back to the client, and send that message down the pipe
                        MessageElement adEl = msg.getElement(ServiceConstants.PIPEADV);
                        PipeAdvertisement backPipeAdv =
                                (PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                                        new MimeMediaType("text/xml"),
                                        adEl.getStream());

                        pipeSvc.createOutputPipe(backPipeAdv, new OutputPipeListener() {
                            public void outputPipeEvent(OutputPipeEvent event) {
                                System.out.println("Server connected back to client");
                                OutputPipe outPipe = event.getOutputPipe();

                                try {
                                    outPipe.send(res);
                                    System.out.println("Sent response message!");
                                } catch (IOException e) {
                                    e.printStackTrace();
                                }
                            }
                        });
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }


    private Message processInput(String jobID, String high, String low) {

        Message m = pipeSvc.createMessage();
        try {
            long startTime = System.currentTimeMillis();
            int[] res = PrimeSearch.findPrimes(Integer.parseInt(low),
                    Integer.parseInt(high));
            long endTime = System.currentTimeMillis();

            //response message elements:
            //jobID, high, low
            //starttime, end time, string containing comma-separated list of primes
            MessageElement id = m.newMessageElement(
                    ServiceConstants.JOBID,
                    new MimeMediaType("text/plain"),
                    jobID.getBytes());

            MessageElement lo = m.newMessageElement(
                    ServiceConstants.LOW_INT,
                    new MimeMediaType("text/plain"),
                    low.getBytes());

            MessageElement hi = m.newMessageElement(
                    ServiceConstants.HIGH_INT,
                    new MimeMediaType("text/plain"),
                    high.getBytes());

            MessageElement startTimeEl = m.newMessageElement(
                    ServiceConstants.STARTTIME, new MimeMediaType("text/plain"),
                    new Long(startTime).toString().getBytes());

            MessageElement endTimeEl = m.newMessageElement(
                    ServiceConstants.ENDTIME, new MimeMediaType("text/plain"),
                    new Long(endTime).toString().getBytes());

            StringBuffer b = new StringBuffer();
            for (int i = 0; i < res.length; i++) {
                b.append(new Integer(res[i]).toString() + ",");
            }
            //for the last one, chop off comma
            b.setLength(b.length() - 1);
            MessageElement primeEl = m.newMessageElement(
                    ServiceConstants.PRIMELIST, new MimeMediaType("text/plain"),
                    b.toString().getBytes());

            m.addElement(id);
            m.addElement(lo);
            m.addElement(hi);
            m.addElement(startTimeEl);
            m.addElement(endTimeEl);
            m.addElement(primeEl);

        } catch (NumberFormatException e) {
            System.out.println("Number format is wrong:" + e.getMessage());
        }

        return m;
    }

    private void startJxta() {

        try {
            group = PeerGroupFactory.newNetPeerGroup();
            discoSvc = group.getDiscoveryService();
            pipeSvc = group.getPipeService();


        } catch (PeerGroupException e) {
            System.out.println("Cannot create Net Peer Group: " + e.getMessage());
            System.exit(-1);
        }

    }


}