student.java

javaP2P技术内幕课程第22课的源代码

package client;

/**
 * This is the client in the Distance Learning example.  It represents the
 * student in the teacher student relationship.
 */
import server.*;
import net.jxta.id.IDFactory;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.credential.AuthenticationCredential;
import net.jxta.document.StructuredDocument;
import net.jxta.document.Document;
import net.jxta.document.StructuredTextDocument;
import net.jxta.document.MimeMediaType;
import net.jxta.membership.Authenticator;
import net.jxta.membership.MembershipService;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupFactory;
import net.jxta.exception.PeerGroupException;
import net.jxta.exception.ProtocolNotSupportedException;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.discovery.DiscoveryService;
import net.jxta.discovery.DiscoveryListener;
import net.jxta.discovery.DiscoveryEvent;
import net.jxta.pipe.PipeService;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.OutputPipeListener;
import net.jxta.pipe.OutputPipeEvent;
import net.jxta.protocol.DiscoveryResponseMsg;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.ResolverQueryMsg;
import net.jxta.protocol.ResolverResponseMsg;
import net.jxta.resolver.ResolverService;
import net.jxta.resolver.QueryHandler;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.document.Advertisement;
import net.jxta.id.ID;
import net.jxta.impl.protocol.ResolverResponse;
import net.jxta.impl.protocol.ResolverQuery;
import java.util.*;
import java.io.*;


public class Student implements Runnable {
    private static final int MAX_TRIES = 5;
    private PeerGroup netPeerGroup;
    private DiscoveryService discoverySvc;
    private PipeAdvertisement pipeTeacherAdv;
    private OutputPipe outputTeacherPipe;
    private PipeAdvertisement pipeStudentAdv;
    private InputPipe inputStudentPipe;

    /**
 * Find an advertisement of a Teacher using JXTA discovery services.
 */
    private void discoverTeacherPipeAdv() throws IOException {
        String name = Teacher.DISTANCE_LEARNING + Teacher.TEACHER_PIPE;
        this.discoverySvc.getRemoteAdvertisements(null, DiscoveryService.ADV, 
                                                  "name", name, 10, null);

        try {
            Thread.sleep(5);
            System.out.print(".");
        } catch (Exception ignore) {
        }

        Enumeration enum = this.discoverySvc.getLocalAdvertisements(
                                   DiscoveryService.ADV, "Name", name);

        while (enum.hasMoreElements()) {
            pipeTeacherAdv = (PipeAdvertisement) enum.nextElement();

            break;
        }
    }

    /**
 * Set up the output pipe to the teacher server.
 */
    private boolean createTeacherPipes() throws IOException {
        PipeService pipeSvc = netPeerGroup.getPipeService();
        this.outputTeacherPipe = pipeSvc.createOutputPipe(this.pipeTeacherAdv, 
                                                          1000);

        return true;
    }

    /**
 * Publish the student advertisement pipes.
 */
    private void publishStudentPipeAdv() throws IOException {
        PipeAdvertisement pipeAdv = null;
        String advType = PipeAdvertisement.getAdvertisementType();

        pipeAdv = (PipeAdvertisement) AdvertisementFactory.newAdvertisement(
                          advType);

        pipeAdv.setPipeID(IDFactory.newPipeID(netPeerGroup.getPeerGroupID()));

        String name = "DL:" + netPeerGroup.getPeerName();
        pipeAdv.setName(name);
        pipeAdv.setType(PipeService.UnicastType);

        discoverySvc.publish(pipeAdv, DiscoveryService.ADV);
        discoverySvc.remotePublish(pipeAdv, DiscoveryService.ADV);
        this.pipeStudentAdv = pipeAdv;
    }

    /**
 * Create an input pipe for inbound messages
 */
    private void createStudentInputPipe() throws IOException {
        PipeService pipeSvc = netPeerGroup.getPipeService();
        this.inputStudentPipe = pipeSvc.createInputPipe(this.pipeStudentAdv);
    }

    /**
 * Wait indefinitely for the teacher to respond to our sent message.
 */
    public Message accept() throws InterruptedException {
        System.out.println("Waiting for the teacher to respond.");

        return inputStudentPipe.waitForMessage();
    }

    /**
 * Send to the Teacher a request for a new series of questions and answers.
 */
    public void requestNewSession() throws IOException {
        Map map = new HashMap();
        map.put("tag", "NEW");
        map.put("course", "whatevers");
        this.sendToTeacher(map);
    }

    /**
 * Get the Student's response from IO.
 */
    private String getUserResponse() throws IOException {
        BufferedReader reader = new BufferedReader(
                                        new InputStreamReader(System.in));

        return reader.readLine();
    }

    /**
 * Utility method to send a response back to the Teacher.
 *
 * @param map Key/value pair to be sent.
 */
    private void sendToTeacher(Map map) throws IOException {
        PipeService pipeSvc = netPeerGroup.getPipeService();
        Message message = pipeSvc.createMessage();
        Set keySet = map.keySet();
        Iterator iterator = keySet.iterator();

        while (iterator.hasNext()) {
            String key = (String) iterator.next();
            String value = (String) map.get(key);

            MessageElement element = message.newMessageElement(key, 
                                                               new MimeMediaType(
                                                                       "text/plain"), 
                                                               new ByteArrayInputStream(
                                                                       value.getBytes()));
            message.addElement(element);
        }

        MessageElement clientElement = message.newMessageElement("client", 
                                                                 new MimeMediaType(
                                                                         "text/plain"), 
                                                                 new ByteArrayInputStream(
                                                                         netPeerGroup.getPeerName()
                                                                                     .getBytes()));
        message.addElement(clientElement);
        this.outputTeacherPipe.send(message);
    }

    /**
 * Processes incoming messages from the teacher object.  Handles questions and
 * the END signal which signifies the end of the question answer session.
 */
    public void processMessage(Message message) throws IOException {
        String tag = getElementValue(message, "tag");
        System.out.println("Received a message from the teacher.");
        System.out.println("tag=" + tag);

        if (tag.equals("QUESTION")) {
            String question = getElementValue(message, "question");
            String number = getElementValue(message, "number");
            String total = getElementValue(message, "total");

            System.out.println("Question [" + number + "/" + total + "]");
            System.out.println(question);
            System.out.print("Your answer: ");

            String reply = getUserResponse();

            // Send back the response to the teacher
            Map map = new HashMap();
            map.put("tag", "ANSWER");
            map.put("ANSWER", reply);
            sendToTeacher(map);
        } else if (tag.equals("END")) {
            // The teacher signals the end of the questions
            // Print out the suggestions
            String suggestions = getElementValue(message, "SUGGESTIONS");
            String response = getElementValue(message, "question");
            System.out.println(response);
            System.out.println(suggestions);
            this.disconnect();
        }
    }

    /**
 * Parses a Message for a specified tag.
 *
 * @param message Contains a message from the Teacher.
 * @param key The key to parse.
 *
 * @returns The value for the key if found.  Null otherwise.
 */
    private String getElementValue(Message message, String key) {
        if (!message.hasElement(key)) {
            return null;
        }

        MessageElement element = message.getElement(key);
        InputStream stream = element.getStream();

        try {
            byte[] bytes = new byte[stream.available()];
            stream.read(bytes);

            return new String(bytes);
        } catch (IOException io) {
            io.printStackTrace();

            return null;
        }
    }

    /**
 * Start up the Net Peer group & get the Discovery Service for that group.
 */
    public void startJxta() throws PeerGroupException, IOException {
        this.netPeerGroup = PeerGroupFactory.newNetPeerGroup();
        this.discoverySvc = netPeerGroup.getDiscoveryService();


        // Flush all local cache information.  We want to start from scratch.
        discoverySvc.flushAdvertisements(null, DiscoveryService.ADV);
        discoverySvc.flushAdvertisements(null, DiscoveryService.PEER);
        discoverySvc.flushAdvertisements(null, DiscoveryService.GROUP);
    }

    /**
 * Establish communications with a Teacher instance through JXTA.
 */
    public boolean connect() throws Exception {
        for (int i = 0; (this.pipeTeacherAdv == null) && (i < MAX_TRIES); i++) {
            this.discoverTeacherPipeAdv();

            try {
                Thread.sleep(5 * 1000);
            } catch (Exception ignore) {
            }
        }

        if (this.pipeTeacherAdv == null) {
            return false;
        }

        try {
            this.createTeacherPipes();
            this.publishStudentPipeAdv();
            this.createStudentInputPipe();

            return true;
        } catch (Exception e) {
            e.printStackTrace();

            return false;
        }
    }

    /**
 * Exit out of the application.
 */
    public void disconnect() {
        System.out.println("bye bye");
        System.exit(0);
    }

    /**
 * Thread to capture the student's responses and process the teacher's messages.
 */
    public void run() {
        try {
            requestNewSession();

            for (;;) {
                try {
                    Message message = this.accept();
                    processMessage(message);
                } catch (InterruptedException ie) {
                }
            }
        } catch (IOException io) {
            io.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
 * Instatiates a Student and launches a thread to allow the student to
 * input answers to questions sent by the Teacher.
 */
    public static void main(String[] argc) throws Exception {
        Student student = new Student();
        student.startJxta();

        if (!student.connect()) {
            System.out.println("Unable to communicate with the teacher.");
            System.out.println("Make sure the Teacher is ready to teach.");
            System.exit(1);
        }

        Thread thread = new Thread(student);
        thread.start();
    }
}