import net.tomp2p.peers.Number160; //导入依赖的package包/类
private void xPut(Entry entry) throws IOException {
    FutureGet fg = peer.get(entry.getID()).getLatest().start()
            .awaitUninterruptibly();
    Pair<Number640, Data> pair = checkVersions(fg.rawData());
    Data newData = new Data(entry);
    Number160 v = pair.element0().versionKey();
    long version = v.timestamp() + 1;
    newData.addBasedOn(v);
    Pair<Number160, Data> pair3 = new Pair<Number160, Data>(new Number160(version,
            newData.hash()), newData);
    FuturePut fp1 = peer.put(entry.getID()).data(Number160.ZERO, pair3.element1().prepareFlag(), pair3.element0()).start().awaitUninterruptibly();
    Pair<Number640, Byte> pair2 = checkVersions(fp1.rawResult());
    FuturePut fp = peer.put(entry.getID())
            .versionKey(pair2.element0().versionKey()).putConfirm()
            .data(new Data()).start().awaitUninterruptibly();
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
private Pair<Number160, Data> getAndUpdate(Directory parentDir, Entry child) throws InterruptedException, ClassNotFoundException,
        IOException {
    Random RND = new Random(42L);
    Pair<Number640, Data> pair = tryToGet(parentDir.getID());
    // we got the latest data
    if (pair != null) {
        // update operation is append
        parentDir.addChild(child.getEntryName(), child.getID(), child.getType());
        Data newData = new Data(parentDir);
        Number160 v = pair.element0().versionKey();
        long version = v.timestamp() + 1;
        newData.addBasedOn(v);
        //since we create a new version, we can access old versions as well
        return new Pair<Number160, Data>(new Number160(version,
                newData.hash()), newData);
    }
    return null;
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
private Pair<Number160, Data> getAndUpdate_remove(Directory parentDir, Entry child) throws InterruptedException, ClassNotFoundException,
        IOException {
    Random RND = new Random(42L);
    Pair<Number640, Data> pair = tryToGet(parentDir.getID());
    // we got the latest data
    if (pair != null) {
    // update operation is append
        parentDir.removeChild(child.getEntryName());
        Data newData = new Data(parentDir);
        Number160 v = pair.element0().versionKey();
        long version = v.timestamp() + 1;
        newData.addBasedOn(v);
    //since we create a new version, we can access old versions as well
        return new Pair<Number160, Data>(new Number160(version,
                newData.hash()), newData);
    }
    return null;
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
private Pair<Number640, Data> tryToGet(Number160 key) {
    Pair<Number640, Data> pair = null;
    for (int i = 0; i < 5; i++) {
        FutureGet fg = peer.get(key).getLatest().start()
                .awaitUninterruptibly();
        // check if all the peers agree on the same latest version, if not
        // wait a little and try again
        pair = checkVersions(fg.rawData());
        if (pair != null) {
            break;
        }
        // something went wrong, have to wait
        try {
            Thread.sleep(RND.nextInt(500));
        } catch (InterruptedException e) {
            //TODO find out what is it
            e.printStackTrace();
        }
    }
    return pair;
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
@Override
public void createDir(String path) throws ClassNotFoundException, FsException, IOException {
    Path subPaths = Paths.get(path);
    int pathLength = subPaths.getNameCount();
    if (pathLength == 0) {
        throw new NotADirectoryException("Don't create a root node like that! Path: "+path);
    }
    Directory parentEntry = getDirectory(subPaths.getParent().toString());

    if (parentEntry.getChild(subPaths.getFileName().toString()) != null) {
        throw new FileExistsException(path);
    }

    Number160 newKey = Number160.createHash(UUID.randomUUID().hashCode());
    Directory newDir = new Directory(newKey, subPaths.getFileName().toString());

    helper.addNewEntry(parentEntry, newDir);
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
@Override
public void createFile(String path) throws ClassNotFoundException, FsException, IOException {

    Path subPaths = Paths.get(path);

    int pathLength = subPaths.getNameCount();
    if (pathLength == 0) {
        throw new NoSuchFileOrDirectoryException("Can not create such file");
    }

    Directory parentEntry = getDirectory(subPaths.getParent().toString());
    if (parentEntry.getChild(subPaths.getFileName().toString()) != null) {
        throw new FileExistsException(path);
    }


    Number160 newKey = Number160.createHash(UUID.randomUUID().hashCode());
    File newFile = new File (newKey, subPaths.getFileName().toString());

    //this is new locking logic, due to fuse constraints we have to associate a file creation with the respective owner
    newFile.setDirtyBit(true);
    newFile.setModifierPeer(peer.peerID());


    helper.addNewEntry(parentEntry, newFile);
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
public static PeerDHT[] createAndAttachPeersDHT(int nr, int port) throws IOException {
    
	Random random = new Random();
	
	PeerDHT[] peers = new PeerDHT[nr];
    for (int i = 0; i < nr; i++) {
        if (i == 0) {
            peers[0] = new PeerBuilderDHT(new PeerBuilder(new Number160(random)).ports(port).start()).start();

        } else {
            peers[i] = new PeerBuilderDHT(new PeerBuilder(new Number160(random)).masterPeer(peers[0].peer()).start()).start();

        }
    }
    return peers;
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
@Before
public void setup() {
    //setup logging to console
    //org.apache.log4j.BasicConfigurator.configure();


    try {
        //initialize network
        peers = createAndAttachPeersDHT(nr, port);
        bootstrap(peers);

        //initialize controller with a peer
        controller = new TreeController(peers[local]);


        //create a root node
        Directory rootDir = new Directory(Number160.ZERO, rootName);
        //upload root into DHT
        Data data = new Data(rootDir);
        FutureDHT futureDHT = peers[3].put(Number160.ZERO).data(data).start();
        futureDHT.awaitUninterruptibly();
    } catch (Exception e) {
        e.printStackTrace();
    }

}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
@Test
public void testRenameEntry() throws ClassNotFoundException, FsException, IOException {
    String oldName = "/entryToRename";
    String newName = "/newName";

    controller.createDir(oldName);

    Number160 entryID = controller.resolvePath(oldName).getID();

    controller.renameEntry(oldName, newName);

    //assert that the entry has the new name
    assertEquals(newName, "/" + getEntryFromID(entryID).getEntryName());

    //assert that the parent also stores the new name
    assertTrue(controller.readDir("/").contains("newName"));
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
public DHTManager initDHT() throws IOException
{
	Random rand = new Random();
	Bindings bind = new Bindings();
	bind.addInterface(Config.getInstance().getNetworkInterface());
	peer = new PeerBuilderDHT(new PeerBuilder(new Number160(rand)).ports(Config.getInstance().getPortDHT()).start()).start();
	
	this.connectToConnectNode();
	/*new IndirectReplication(peer).start();
	
	InetAddress address = Inet4Address.getByName(Config.getInstance().getConnectNode());
	FutureDiscover futureDiscover = peer.peer().discover().inetAddress(address).ports(Config.getInstance().getPortDHT()).start();
	futureDiscover.awaitUninterruptibly();
	FutureBootstrap futureBootstrap = peer.peer().bootstrap().inetAddress(address).ports(Config.getInstance().getPortDHT()).start();
	futureBootstrap.awaitUninterruptibly();*/
	
	return this;
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
/**
 * Retrieves the social record from the DHT.
 *
 * @param key
 * @return the social record
 * @throws ClassNotFoundException
 * @throws IOException
 * @throws GUIDNotFoundException 
 */
public String get(String key) throws GUIDNotFoundException
{
	try
	{
		FutureGet futureGet = peer.get(Number160.createHash(key)).start();
		futureGet.awaitUninterruptibly();
		
		// TODO: use non-blocking?
		if(futureGet.isSuccess() && futureGet.data() != null)
		{
			return futureGet.data().object().toString();
		}
		else
		{
			throw new GUIDNotFoundException("GUID not found");
		}
		
		//return null; // TODO: decide on sentinel value
	}
	catch(ClassNotFoundException | IOException e)
	{
		throw new GUIDNotFoundException("GUID not found");
	}
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
public void updateOverlay(PeerDHT peer)	throws OverlayException {
	log.info("Updating UNaDa information inside DHT.");
	updateUnadaInfo(peer);
	manager.resolveGeoLocation(uNaDaInfo);
	FutureRemove remove = peer.remove(Number160.createHash(uNaDaInfo.getUnadaID())).all().start();
	try {
		remove.await();
	} catch (InterruptedException e1) {
		log.error("Failed to delete old contents.", e1);
	}
	
	FuturePut futurePut;
	
	try {
		
		log.debug("Updating Unada with address: {}, tcp port {}, udp port {}.", uNaDaInfo.getUnadaAddress(), uNaDaInfo.getTcpPort(), uNaDaInfo.getUdpPort());
		futurePut = peer.put(Number160.createHash(uNaDaInfo.getUnadaID())).data(new Data(uNaDaInfo).ttlSeconds(UNADA_INFO_TTLs + 30)).start();
		futurePut.awaitUninterruptibly();
		if(!futurePut.isSuccess()){
			throw new OverlayException("Put for UNaDa info failed.");
		}
	} catch (IOException e) {
		throw new OverlayException("Unexpected Exception:", e);
	}
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
public void advertiseContent(long... contentID) {
	log.info("Advertising contents into DHT: ", contentID);
	deleteContents(contentID);
	for(long id : contentID){
		try {
			FuturePut put = peer.put(Number160.createHash(id)).data(Number160.createHash(uNaDaInfo.getUnadaID()), new Data(uNaDaInfo).ttlSeconds(UNADA_INFO_TTLs + 30)).start();
			put.awaitUninterruptibly();
			if(put.isSuccess()){
				log.debug("{} - Successfully advertised conten {} to DHT.",uNaDaInfo.getUnadaID() , id);
			}else{
				log.warn("{} - Failed to advertised conten {} to DHT.", uNaDaInfo.getUnadaID(), id);
			}
		} catch (IOException e) {
			log.error("{} - Failed while advertising conten {}.", uNaDaInfo.getUnadaID(), contentID, e);
		}
		manager.queryProviders(id, null);
	}
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
/**
     * Called when the job owner has been notified that a certain result has been uploaded.
     * @param replicaID ID of the replica who's result was uploaded
     */
    private void resultUploaded(final ReplicaID replicaID){
        System.out.println("Replica was completed: "+replicaID);

        final Number160 resultKey = replicaManager.getReplicaResultKey(replicaID);
//        System.out.println("\tResultKey: "+resultKey);

        client.addListener(new OperationFinishedListener(client, resultKey, CommandWord.GET) {
            @Override
            protected void operationFinished(Operation operation) {
                if (operation.isSuccess()) {
//                    System.out.println("RESULT RAW: "+operation.getResult().toString());
                    Data resultData = (Data) operation.getResult();

                    byte[] resultArray = resultData.getData();
                    System.out.println("Result downloaded successfully, \n\tresult holds "+resultArray.length+" bytes.");
                    replicaManager.replicaFinished(replicaID, resultArray);
                } else {
                    System.out.println("DownloadOperation failed! " + operation.getErrorCode()
                            + "\n\t" + operation.getReason());
                }
            }
        });
        client.get(resultKey, client.getID());

    }
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
public void issueRequest(final Request r) throws IOException {
	String[] temp = { "" + r.getId(), r.getFingerprint() };
	FutureDHT fdht = p.send(new Number160(new Random())).setObject(temp)
			.start();

	fdht.addListener(new BaseFutureAdapter<FutureDHT>() {

		@Override
		public void operationComplete(FutureDHT f) throws Exception {
			if (f.isFailed()) {
				Logger.getLogger(PeerManager.class.getName()).log(
						Level.WARN,
						"Request " + r.getId() + "from User " + r.getUser().getName()
								+ " failed.");
			} else {
				Logger.getLogger(PeerManager.class.getName()).log(
						Level.INFO,
						"Request " + r.getId() + r.getId() + "from User "
								+ r.getUser().getName() + " successfully issued.");
			}

		}

	});
}
 

import net.tomp2p.peers.Number160; //导入依赖的package包/类
/**
 * Finish the future and set the keys that have been stored. Success or failure is determined if the communication
 * was successful. This means that we need to further check if the other peers have denied the storage (e.g., due to
 * no storage space, no security permissions). Further evaluation can be retrieved with {@link #getAvgStoredKeys()}
 * or if the evaluation should be done by the user, use {@link #getRawKeys()}.
 * 
 * @param domainKey
 *            The domain key
 * @param locationKey
 *            The location key
 * @param rawKeys
 *            The keys that have been stored with information on which peer it has been stored
 * @param rawKeys480
 *            The keys with locationKey and domainKey Flag if the user requested putIfAbsent
 */
public void setStoredKeys(final Number160 locationKey, final Number160 domainKey,
        final Map<PeerAddress, Collection<Number480>> rawKeys480) {
    synchronized (lock) {
        if (!setCompletedAndNotify()) {
            return;
        }
        this.rawKeys480 = rawKeys480;
        this.locationKey = locationKey;
        this.domainKey = domainKey;
        final int size = rawKeys480 == null ? 0 : rawKeys480.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.Number160; //导入依赖的package包/类
/**
 * Finishes the future and set the digest information that have been received.
 * 
 * @param domainKey
 *            The domain key
 * @param locationKey
 *            The location key
 * @param rawDigest
 *            The hashes of the content stored with information from which peer it has been received.
 */
public void setReceivedDigest(final Number160 locationKey, final Number160 domainKey,
        final Map<PeerAddress, DigestResult> rawDigest) {
    synchronized (lock) {
        if (!setCompletedAndNotify()) {
            return;
        }
        this.locationKey = locationKey;
        this.domainKey = domainKey;
        this.rawDigest = rawDigest;
        final int size = rawDigest.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.Number160; //导入依赖的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.Number160; //导入依赖的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.Number160; //导入依赖的package包/类
@Test
public void testBootstrap4() throws Exception {
    final Random rnd = new Random(42);
    Peer master = null;
    Peer slave = null;
    try {
        master = new PeerMaker(new Number160(rnd)).ports(4001).makeAndListen();
        slave = new PeerMaker(new Number160(rnd)).ports(4002).makeAndListen();
        BaseFuture res = slave.ping().setPort(4001).setBroadcast().start();
        res.awaitUninterruptibly();
        Assert.assertEquals(true, res.isSuccess());
    } finally {
        if (master != null) {
            master.shutdown().await();
        }
        if (slave != null) {
            slave.shutdown().await();
        }
    }
}