import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
private static <K> Pair<Number640, K> checkVersions(
        Map<PeerAddress, Map<Number640, K>> rawData) {
    Number640 latestKey = null;
    K latestData = null;
    for (Map.Entry<PeerAddress, Map<Number640, K>> entry : rawData
            .entrySet()) {
        if (latestData == null && latestKey == null) {
            latestData = entry.getValue().values().iterator().next();
            latestKey = entry.getValue().keySet().iterator().next();
        } else {
            if (!latestKey.equals(entry.getValue().keySet().iterator()
                    .next())
                    || !latestData.equals(entry.getValue().values()
                    .iterator().next())) {
                return null;
            }
        }
    }
    return new Pair<Number640, K>(latestKey, latestData);
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@RequestMapping(method = RequestMethod.GET, produces = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<String> index() throws URISyntaxException
{
	LOGGER.info("Incoming request: GET /");
	List<PeerAddress> AllNeighbors = DHTManager.getInstance().getAllNeighbors();
	
	JSONArray connectedNodes = new JSONArray();
	
	for (PeerAddress neighbor : AllNeighbors) {
		connectedNodes.put(neighbor.inetAddress().getHostAddress());
	}
	
	JSONObject version = new JSONObject();
	version.put("date", Config.getInstance().getVersionDate());
	version.put("version", Config.getInstance().getVersionName());
	version.put("build", Config.getInstance().getVersionNumber());
	
	JSONObject response = new JSONObject();
	response.put("Code", 200);
	response.put("Description", "OK");
	response.put("Value", "");
	response.put("version", version);
	response.put("connectedNodes", connectedNodes);
	
	return new ResponseEntity<String>(response.toString(), HttpStatus.OK);
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@Override
public void run() {
    DeceitfulNetworkUtils.bootstrap(peer, "narrens.olf.sgsnet.se", 4001, new OnReplyCommand() {
        @Override
        public void execute(Object replyMessageContent) {
            System.out.println("\tBootstrap done");

            PeerAddress jobOwner = (PeerAddress) replyMessageContent;
            if(jobOwner == null){
                peer.shutdown();
            } else {
                taskPasserDeny.requestWork(jobOwner);
            }
        }
    });
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * Send a message to this Peer and expect an answer
 *
 * @param receiver peer
 * @param message message
 * @param onReturn what you will do when it answers
 */
protected void sendRequest(final PeerAddress receiver, Serializable message, final OnReplyCommand onReturn){
    SendBuilder sendBuilder = peer.send(receiver.getID());

    final NetworkMessage networkMessage = new NetworkMessage(message, NetworkMessage.Type.REQUEST);

    FutureDHT futureDHT = sendBuilder.setObject( networkMessage.encrypt() ).setRequestP2PConfiguration(requestConfiguration).start();
    futureDHT.addListener(new BaseFutureAdapter<FutureDHT>() {
        @Override
        public void operationComplete(FutureDHT future) throws Exception {
            if(!future.isSuccess()){
                System.out.println("Error sending " + networkMessage.toString());
                System.out.println("WHY: "+future.getFailedReason());
                return;
            }

            System.out.println("Success sending " + networkMessage.toString() + " to " + print(receiver));
            for(PeerAddress address : future.getRawDirectData2().keySet()){
                Object answer = future.getRawDirectData2().get(address);
                onReturn.execute(answer);
                System.out.println(print(address)+" answered with "+answer);
            }
        }
    });
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * Send message to a peer without expecting something in reply
 * @param receiver peer
 * @param message message
 */
protected void sendNoReplyMessage(PeerAddress receiver, Serializable message){
    SendBuilder sendBuilder = peer.send(receiver.getID());

    final NetworkMessage networkMessage = new NetworkMessage(message, NetworkMessage.Type.NO_REPLY);

    FutureDHT futureDHT = sendBuilder.setObject( networkMessage.encrypt() ).setRequestP2PConfiguration(requestConfiguration).start();
    futureDHT.addListener(new BaseFutureAdapter<FutureDHT>() {
        @Override
        public void operationComplete(FutureDHT future) throws Exception {
            if(!future.isSuccess()){
                System.out.println("Error sending " + networkMessage.toString());
                System.out.println("WHY: "+future.getFailedReason());
                return;
            }
            //Disabled for demo
            System.out.println("Success sending " + networkMessage.toString());
        }
    });
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
synchronized protected void handleNoReply(PeerAddress sender, Object messageContent) {
    TaskMessage taskMessage = TaskMessage.check(messageContent);

    switch (taskMessage.getType()){
        case RESULT_UPLOADED:
            resultUploaded((ReplicaID) taskMessage.getActualContent());
            break;
        case TASK_FAIL:
            FailMessage failMessage = (FailMessage) taskMessage.getActualContent();
            WorkerID worker = taskMessage.getSenderID();
            //TODO check reputation as well?
            //TODO handle in ReplicaManager instead
            if(replicaManager.isWorkerAssignedReplica(worker, failMessage.getReplicaID())){
                System.out.println("My task failed! Reason: "+failMessage.getReason());
                replicaManager.replicaFailed(failMessage.getReplicaID());
            } else {
                System.out.println("Warning! A worker node reported a failure in a task it was not participating in...");
                workerReputationManager.reportWorker(worker);
            }
            break;
        default:
            throw new UnsupportedOperationException("Unsupported request: "+taskMessage.getType());
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@Override
public void peerInserted(PeerAddress peerAddress) {

    Boolean bootstrap = false;

    for(String [] s: bootstrapNodes) {
        if(s[0].equals(peerAddress.getInetAddress().getHostAddress()) ||
                s[0].equals(peerAddress.getInetAddress().getCanonicalHostName())) {
            if(peerAddress.portTCP() == Integer.parseInt(s[1])) {
                bootstrap = true;
            }
        }
    }

    if(bootstrap) {
        return;
    }

    Boolean added = fileNeighbours.add(peerAddress);

    if(added) {
        writeNeighbours(peerAddress);
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@Override
public void peerUpdated(PeerAddress peerAddress) {

    Boolean bootstrap = false;

    for(String [] s: bootstrapNodes) {
        if(s[0].equals(peerAddress.getInetAddress().getHostAddress()) ||
                s[0].equals(peerAddress.getInetAddress().getCanonicalHostName())) {
            if(peerAddress.portTCP() == Integer.parseInt(s[1])) {
                bootstrap = true;
            }
        }
    }

    if(bootstrap) {
        return;
    }


    fileNeighbours.remove(peerAddress);
    fileNeighbours.add(peerAddress);

    updateNeighbour(peerAddress);

}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
protected NavigableMap<Number160, Set<Number160>> buildDigest(Map<PeerAddress, DigestResult> rawDigest) {
	NavigableMap<Number160, Set<Number160>> digestMap = new TreeMap<Number160, Set<Number160>>();
	if (rawDigest == null) {
		return digestMap;
	}
	for (PeerAddress peerAddress : rawDigest.keySet()) {
		NavigableMap<Number640, Collection<Number160>> tmp = rawDigest.get(peerAddress).keyDigest();
		if (tmp == null || tmp.isEmpty()) {
			// ignore this peer
		} else {
			for (Number640 key : tmp.keySet()) {
				for (Number160 bKey : tmp.get(key)) {
					if (!digestMap.containsKey(key.versionKey())) {
						digestMap.put(key.versionKey(), new HashSet<Number160>());
					}
					digestMap.get(key.versionKey()).add(bKey);
				}
			}
		}
	}
	return digestMap;
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
private void preHandleMessage(Message2 message, TrackerStorage trackerStorage, PeerAddress referrer,
        Number160 locationKey, Number160 domainKey) throws IOException, ClassNotFoundException {
    // Since I might become a tracker as well, we keep this information
    // about those trackers.
    TrackerData tmp = message.getTrackerData(0);
    // no data found
    if (tmp == null || tmp.size() == 0) {
        return;
    }
    for (Map.Entry<PeerAddress, Data> trackerData : tmp.getPeerAddresses().entrySet()) {
        // we don't know the public key, since this is not first hand
        // information.
        // TTL will be set in tracker storage, so don't worry about it here.
        trackerStorage.putReferred(locationKey, domainKey, trackerData.getKey(), referrer,
                trackerData.getValue(), ReferrerType.MESH);
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * Start the sender and wait forever.
 * 
 * @throws InterruptedException .
 * @throws IOException .
 */
@Test
public void startSender() throws IOException, InterruptedException {

    final int maxPeers = 10;
    Peer[] peers = createPeers(PORT, maxPeers, "sender");
    for (Peer peer : peers) {
        peer.bootstrap().setInetAddress(InetAddress.getByName(ADDR)).setPorts(PORT).start().awaitUninterruptibly();
        peer.setObjectDataReply(new ObjectDataReply() {
            @Override
            public Object reply(final PeerAddress sender, final Object request) throws Exception {
                System.out.println("wrong!!!!");
                return "wrong!!!!";
            }
        });
    }
    Number160 keyForID = Number160.createHash("key");
    Msg message = new Msg();

    System.out.println(String.format("Sending message '%s' to key '%s' converted to '%s'", message.getType(),
            "key", keyForID.toString()));
    FutureDirect futureDHT = peers[0].send(keyForID).setObject(message).setRequestP2PConfiguration(REQ).start();
    futureDHT.awaitUninterruptibly();
    System.out.println("got: " + futureDHT.getObject());
    Thread.sleep(Long.MAX_VALUE);
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
private void notifyMasterPeer(List<PeerAddress> peerList, BaseNotificationMessageFactory messageFactory, String userId,
		PublicKey publicKey) {
	logger.debug("Notifying master peer of user {}", userId);
	boolean success = false;
	while (!success && !peerList.isEmpty()) {
		PeerAddress initial = NetworkUtils.choseFirstPeerAddress(peerList);
		BaseDirectMessage msg = messageFactory.createHintNotificationMessage(initial, userId);
		try {
			send(msg, publicKey);
			success = true;
		} catch (SendFailedException e) {
			if (!peerList.isEmpty()) {
				logger.error("Initial peer of user '{}' was offline. Try next in line.", userId, e);
				peerList.remove(0);
			}
		}
	}

	if (success) {
		logger.debug("Successfully notified the initial peer of user '{}' that it should check its UP tasks.", userId);
	} else {
		logger.info("All clients of user '{}' are currently offline or unreachable.", userId);
	}
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@Override
protected Void doExecute() throws InvalidProcessStateException {
	logger.debug("Starting to get all locations from the users to be notified.");
	Map<String, List<PeerAddress>> allLocations = new HashMap<String, List<PeerAddress>>();

	// iterate over all users and get the locations of them
	for (String userId : context.consumeUsersToNotify()) {
		BaseNetworkContent content = get(userId, H2HConstants.USER_LOCATIONS);
		if (content == null) {
			allLocations.put(userId, new ArrayList<PeerAddress>());
		} else {
			Locations currentLoc = (Locations) content;
			List<PeerAddress> addresses = new ArrayList<PeerAddress>(currentLoc.getPeerAddresses());
			allLocations.put(userId, addresses);
		}
	}

	// done with all locations
	logger.debug("Sending notifications to {} users: ", allLocations.size(), allLocations.keySet().toArray());
	context.setAllLocations(allLocations);

	return null;
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * Update responsibilities. This happens for a put / add.
 * 
 * @param locationKey
 *            The location key.
 */
public void updateAndNotifyResponsibilities(final Number160 locationKey) {
    if (!isReplicationEnabled()) {
        return;
    }
    PeerAddress closest = closest(locationKey);
    if (closest.getPeerId().equals(selfAddress.getPeerId())) {
        if (replicationStorage.updateResponsibilities(locationKey, closest.getPeerId())) {
            // I am responsible for this content
            notifyMeResponsible(locationKey);
        }
    } else {
        if (replicationStorage.updateResponsibilities(locationKey, closest.getPeerId())) {
            // notify that someone else is now responsible for the
            // content with key responsibleLocations
            notifyOtherResponsible(locationKey, closest);
        }
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
public static void routing(Number160 key, Peer[] peers, int start) {
    System.out.println("routing: searching for key " + key);
    NavigableSet<PeerAddress> pa1 = new TreeSet<PeerAddress>(PeerMap.createComparator(key));
    NavigableSet<PeerAddress> queried = new TreeSet<PeerAddress>(PeerMap.createComparator(key));
    Number160 result = Number160.ZERO;
    Number160 resultPeer = new Number160("0xd75d1a3d57841fbc9e2a3d175d6a35dc2e15b9f");
    int round = 0;
    while (!resultPeer.equals(result)) {
        System.out.println("round " + round);
        round++;
        pa1.addAll(peers[start].getPeerBean().peerMap().getAll());
        queried.add(peers[start].getPeerAddress());
        System.out.println("closest so far: " + queried.first());
        PeerAddress next = pa1.pollFirst();
        while (queried.contains(next)) {
            next = pa1.pollFirst();
        }
        result = next.getPeerId();
        start = findNr(next.getPeerId().toString(), peers);
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * The first peer is the initiator. This peer that wants to start the broadcast will send it to all its neighbors.
 * Since this peer has an interest in sending, it should also work more than the other peers.
 * 
 * @param messageKey
 *            The key of the message
 * @param dataMap
 *            The data map to send around
 * @param hopCounter
 *            The number of hops
 * @param isUDP
 *            Flag if message can be sent with UDP
 */
private void firstPeer(final Number160 messageKey, final Map<Number480, Data> dataMap, final int hopCounter,
        final boolean isUDP) {
    final List<PeerAddress> list = peer.getPeerBean().peerMap().getAll();
    for (final PeerAddress peerAddress : list) {
        FutureChannelCreator frr = peer.getConnectionBean().reservation().create(isUDP?1:0, isUDP?0:1);
                frr.addListener(new BaseFutureAdapter<FutureChannelCreator>() {
                    @Override
                    public void operationComplete(final FutureChannelCreator future) throws Exception {
                        if (future.isSuccess()) {
                            BroadcastBuilder broadcastBuilder = new BroadcastBuilder(peer, messageKey);
                            broadcastBuilder.dataMap(dataMap);
                            broadcastBuilder.hopCounter(hopCounter + 1);
                            FutureResponse futureResponse = peer.getBroadcastRPC().send(peerAddress, broadcastBuilder, 
                                    future.getChannelCreator(), broadcastBuilder);
                            LOG.debug("1st broadcast to {}", peerAddress);
                            Utils.addReleaseListener(future.getChannelCreator(), futureResponse);
                        }
                    }
                });
    }
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
private FutureWrappedBootstrap<FutureBootstrap> bootstrapPing(PeerAddress address) {
    final FutureWrappedBootstrap<FutureBootstrap> result = new FutureWrappedBootstrap<FutureBootstrap>();
    final FutureResponse tmp = (FutureResponse) peer.ping().setPeerAddress(address).setTcpPing().start();
    tmp.addListener(new BaseFutureAdapter<FutureResponse>() {
        @Override
        public void operationComplete(final FutureResponse future) throws Exception {
            if (future.isSuccess()) {
                peerAddress = future.getResponse().getSender();
                bootstrapTo = new ArrayList<PeerAddress>(1);
                bootstrapTo.add(peerAddress);
                result.setBootstrapTo(bootstrapTo);
                result.waitFor(bootstrap());
            } else {
                result.setFailed("could not reach anyone with bootstrap");
            }
        }
    });
    return result;
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * Finish the future and set the keys and data that have been received.
 * 
 * @param domainKey
 *            The domain key
 * @param locationKey
 *            The location key
 * @param rawData
 *            The keys and data that have been received with information from which peer it has been received.
 */
public void setReceivedData(final Number160 locationKey, final Number160 domainKey,
        final Map<PeerAddress, Map<Number480, Data>> rawData) {
    synchronized (lock) {
        if (!setCompletedAndNotify()) {
            return;
        }
        this.locationKey = locationKey;
        this.domainKey = domainKey;
        this.rawData = rawData;
        final int size = rawData.size();
        this.minReached = size >= min;
        this.type = size > 0 ? FutureType.OK : FutureType.FAILED;
        this.reason = size > 0 ? "Minimun number of results reached" : "Expected >0 result, but got " + size;
    }
    notifyListerenrs();
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public List<PeerAddress> getPlayers(InetAddress address) {
	if (address == null) {
		return null;
	} else {
		List<PeerAddress> l = new ArrayList<PeerAddress>();
		if (peer.getPeerAddress().getInetAddress().equals(address)) {
			l.add(peer.getPeerAddress());
		}
		PeerMap pm = peer.getPeerBean().getPeerMap();
		for (PeerAddress pa : pm.getAll()) {
			if (pa.getInetAddress().equals(address)) {
				l.add(pa);
			}
		}
		return l;
	}
}
 

import net.tomp2p.peers.PeerAddress; //导入依赖的package包/类
@Test
public void testBootstrap5() throws Exception {
    final Random rnd = new Random(42);
    Peer peer = null;
    try {
        peer = new PeerMaker(new Number160(rnd)).ports(4000).makeAndListen();
        PeerAddress pa = new PeerAddress(new Number160(rnd), "192.168.77.77", 4000, 4000);
        FutureBootstrap tmp = peer.bootstrap().setPeerAddress(pa).start();
        tmp.awaitUninterruptibly();
        Assert.assertEquals(false, tmp.isSuccess());
    } finally {
        if (peer != null) {
            peer.shutdown().await();
        }
    }
}