hungrypeer.java

JXTA技术手册 书上3.4.5章的源代码

                // If we found some advertisements in the cache,
                // add them to the list
                if (ae != null && ae.hasMoreElements()) {
                    // Reset count and RestoPeerAdvs as we have
                    // just retrieved all advertisements, including
                    // previously cached ones
                    found = 0;
                    restoPeerAdvs.removeAllElements();
                    while (ae.hasMoreElements()) {
                        restoPeerAdvs.addElement(ae.nextElement());
                        ++found;
                    }
                    if (found > 1)
                        break;     // Want to find at least two
                }

                // Did not find enough advertisement in the cache.
                // Send a remote discovery request to search for
                // more RestoPeer advertisements
                disco.getRemoteAdvertisements(null,
                             DiscoveryService.ADV,
                             "name", "RestoNet:RestoPipe:*", 5, null);

                // Give the peers a chance to respond
                try {
                    Thread.sleep(timeout);
                } catch (InterruptedException e) {}
            } catch (IOException e){
                // Found nothing! Move on
            }
        }
        // Completed RestoPeer Discovery
        System.out.println("Found " + found + " RestoPeers(s)");
    }

    // Method to connect and open output pipes to all the
    // RestoPeers that we have discovered. Each RestoPeer is
    // identified by its unique RestoPeer pipe advertisement.
    private void connectToRestoPeers() {
        // Enumerate all the RestoPeer pipe advertisments we have discovered
        // and attempt to connect a pipe which each of them
        for (Enumeration en = restoPeerAdvs.elements();
             en.hasMoreElements();) {

            PipeAdvertisement padv = (PipeAdvertisement) en.nextElement();
            try {
                System.out.println(
                    "Attempting to connect to discovered RestoPeer");

                // Create an output pipe connection to the RestoPeer
                OutputPipe pipeOut = pipes.createOutputPipe(padv,
                                                            rtimeout);

                // Check if we have a connected pipe
                if (pipeOut == null) {
                    // Failed; go to next RestoPeer
                    System.out.println(
                        "Failure to connect to RestoPeer Pipe:" +
                        padv.getName());
                    continue;
                }

                // Save the output pipe
                restoPeerPipes.addElement(pipeOut);
                System.out.println("Connected pipe to " + padv.getName());
            } catch (Exception e) {
                // Error during connection go to next RestoPeer
                System.out.println("RestoPeer may not be there anymore:" +
                                    padv.getName());
                 continue;
            }
        }
    }

    // Send an auction request for French Fries to all the RestoPeer
    // pipes we have successfully connected
    private void sendFriesAuctionRequests() {
        // Enumerate all the RestoPeer pipe connections we have successfully
        // connected to
        for (Enumeration en = restoPeerPipes.elements();
            en.hasMoreElements();) {
            OutputPipe op = (OutputPipe) en.nextElement();
            try {
                // Construct the request document
                StructuredDocument request  =
                    StructuredDocumentFactory.newStructuredDocument(mimeType,
                    "RestoNet:Request");

                // Fill up the Fries auction request argument
                Element re;
                re = request.createElement("Name", myIdentity);
                request.appendChild(re);
                re = request.createElement("Fries", friesRequest);
                request.appendChild(re);

                // Create the pipe message to send
                Message msg = pipes.createMessage();

                // Fill the first message element which is the HungryPeer
                // pipe advertisement return address. We need this
                // so RestoPeers can respond to us
                msg.addElement(msg.newMessageElement("HungryPeerPipe",
                         mimeType, myAdv.getDocument(mimeType).getStream()));

                // Fill the second message element, which is
                // the fries request. Insert the document
                // in the message
                msg.addElement(msg.newMessageElement("Request",
                         mimeType, request.getStream()));

                // Send the auction message to the RestoPeer
                op.send(msg);
                System.out.println("Sent Fries Auction Request ("
                    + friesRequest + ") to connected peers");
            } catch (Exception ex) {
                // Error sending auction request
                System.out.println(
                    "Failed to send auction request to RestoPeer");
            }
        }
    }

    // Receive bid requests from RestoPeers on the
    // HungryPeer listening pipe
    private void receiveFriesBids() {
        // Continue until we get all answers
        while (true) {
            Message msg = null;      // Pipe message received
            String price = null;     // Fries price bid
            String brand = null;     // RestoPeer name which offers the bid
            String specials = null;  // Specials offer bid
            InputStream ip = null;   // Input stream to read message element
            StructuredDocument bid = null; //Bid document received

            try {
                // Wait for a bid message to arrive from a RestoPeer
                // Will block until a message arrive
                msg = myPipe.waitForMessage();

                // Check if the message is valid
                if (msg == null) {
                    if (Thread.interrupted()) {
                        // We have been asked to stop
                        System.out.println(
                            "Abort Receiving bid loop interrupted");
                        myPipe.close(); // Close the Pipe
                        return;
                    }
                }
            } catch (Exception ex) {
                // Error in receiving message
                myPipe.close();
                System.out.println("Abort Receiving Error receiving bids");
                return;
            }

            // We got a message from a RestoPeer.
            // Extract and display infomation about the bid received.
            try {
                // Extract the Bid document from the message
                ip = msg.getElement("Bid").getStream();
                bid = StructuredDocumentFactory.newStructuredDocument(
                            mimeType, ip);

                // Parse the document to extract bid information
                Enumeration enum = bid.getChildren();
                while (enum.hasMoreElements()) {
                    Element element = (Element) enum.nextElement();
                    String attr = (String) element.getKey();
                    String value = (String) element.getValue();
                    if (attr.equals("Price")) {
                        price = value;
                        continue;
                    }
                    if (attr.equals("Brand")) {
                        brand = value;
                        continue;
                    }
                    if (attr.equals("Specials")) {
                        specials = value;
                        continue;
                    }
                }

                // We got a valid bid. Print it.
                System.out.println("Received Fries Bid from RestoPeers (" +
                           brand + ") at a Price ($" + price +
                           ") \nRestoPeers Special (" + specials + ")");
            } catch (Exception e) {
                // Broken content
                System.out.println("Error extracting bid from the message");
                continue;
            }
        }
    }

    private void authenticateAndJoin(PeerGroup pg) {
        try {
            StructuredDocument creds = null;
            AuthenticationCredential authCred =
                new AuthenticationCredential(pg, null, creds);
            MembershipService m =
                (MembershipService) pg.getMembershipService();
            Authenticator auth = m.apply(authCred);
            PasswdMembershipService.PasswdAuthenticator pwAuth =
                (PasswdMembershipService.PasswdAuthenticator) auth;
            String user = "hungrypeer";
            String passwd = "password";
            System.out.println("setting user to " + user + ":" + passwd);
            pwAuth.setAuth1Identity(user);
            pwAuth.setAuth2_Password(passwd);
            if (!auth.isReadyForJoin()) {
                System.err.println("Authentication failed");
                throw new IllegalArgumentException("Can't authenticate");
            }
            m.join(auth);
        } catch (Exception e) {
            System.err.println("Authentication failed");
            throw new IllegalArgumentException(e.toString());
        }
    }
}