jxtahelper.java

jxta聊天和文件获取 可以和列表中的peer通信

			if (LOG.isEnabledFor(Level.DEBUG)) {
				LOG.debug(e);
			}
			return;
		}
	}

	public void sendMsg(String msgString) {
		try {
			Message msg = new Message();
			msg.addMessageElement(msgPipeName, new StringMessageElement(
					msgPipeName, msgString, null));
			System.out.println("Sending: " + msgString);

			OutputPipe outpipe = pipes.createOutputPipe(msgOutPipeAdv, 3000);
			outpipe.send(msg);
		} catch (Exception ie) {
			ie.printStackTrace();
		}
	}

	private void prepareDataSocket() {
		/**
		 * advertise that we have a data socket to listen on binary data
		 */
		ModuleClassAdvertisement mcadv = (ModuleClassAdvertisement) AdvertisementFactory
				.newAdvertisement(ModuleClassAdvertisement
						.getAdvertisementType());
		mcadv.setName("JXTAMOD:" + dataSocketName);
		mcadv
				.setDescription("Fileshare Module Advertisement for binary data socket.");
		ModuleClassID mcID = IDFactory.newModuleClassID();
		mcadv.setModuleClassID(mcID);

		/**
		 * remote publishing is of course needed so that the others know, we are
		 * there and offer a socket local publishing is needed, otherwise the
		 * local runtime is not informed when someone wants to reach us on the
		 * socket
		 */
		try {
			discovery.publish(mcadv);
		} catch (Exception e) {
			e.printStackTrace();
		}
		discovery.remotePublish(mcadv);

		ModuleSpecAdvertisement msAdv = (ModuleSpecAdvertisement) AdvertisementFactory
				.newAdvertisement(ModuleSpecAdvertisement
						.getAdvertisementType());
		msAdv.setName("JXTASPEC:" + dataSocketName);
		msAdv.setVersion("Version 0.9");
		msAdv.setCreator("Christian Sell & Martin Knechtel");
		msAdv.setModuleSpecID(IDFactory.newModuleSpecID(mcID));
		msAdv.setSpecURI("http://www.knechtel.eu/filesharing/datasocket");

		PipeAdvertisement pipeAdv = null;
		try {
			InputStream is = getClass().getResourceAsStream("./datasocket.adv");
			pipeAdv = (PipeAdvertisement) AdvertisementFactory
					.newAdvertisement(MimeMediaType.XMLUTF8, is);
			is.close();
		} catch (IOException ioe) {
			System.err.println("Faild to read/parse pipe advertisement");
			return;
		}
		/**
		 * each time new pipe-id, otherwise the peer sends to itself if he holds
		 * the same pipe id listening
		 */
		pipeAdv.setPipeID(IDFactory.newPipeID(netPeerGroup.getPeerGroupID()));
		msAdv.setPipeAdvertisement(pipeAdv);

		if (6 < DEBUGLEVEL) {
			// display the advertisement as a plain text document.
			StructuredTextDocument doc = (StructuredTextDocument) msAdv
					.getDocument(MimeMediaType.XMLUTF8);
			try {
				StringWriter out = new StringWriter();
				doc.sendToWriter(out);
				System.out.println(out.toString());
				out.close();
			} catch (IOException ioe) {
			}
		}

		try {
			discovery.publish(msAdv);
		} catch (Exception e) {
			e.printStackTrace();
		}
		discovery.remotePublish(msAdv);

		try {
			dataInSocket = new JxtaServerSocket(netPeerGroup, pipeAdv);
		} catch (IOException e1) {
			e1.printStackTrace();
		}

		/**
		 * discover listening sockets of our neighbours, choose one of them
		 */
		System.out
				.print("searching for a listening socket of a peer for binary data");
		Enumeration en = null;
		boolean pipefound = false;
		while (!pipefound) {
			try {
				// search remotely and fill our cache
				discovery.getRemoteAdvertisements(null, DiscoveryService.ADV,
						"Name", "JXTASPEC:" + dataSocketName, 10, null);

				// let's look in our local cache to see
				// if we have it!
				en = discovery.getLocalAdvertisements(DiscoveryService.ADV,
						"Name", "JXTASPEC:" + dataSocketName);

				while (en != null && en.hasMoreElements()) {
					ModuleSpecAdvertisement mdsadv = (ModuleSpecAdvertisement) en
							.nextElement();
					try {
						// we can find the pipe to connect to the service
						// in the advertisement.
						dataOutPipeAdv = mdsadv.getPipeAdvertisement();

						// is the found pipe another than we ourselves
						// advertised?
						String ourInputPipeId = pipeAdv.getPipeID().toString();
						String discoveredInputPipeID = dataOutPipeAdv
								.getPipeID().toString();
						if (!ourInputPipeId.equals(discoveredInputPipeID)) {
							pipefound = true;
							break; // we just take the first foreign pipe
							// later: let user choose to connect to which peer
						}
					} catch (Exception ex) {
						ex.printStackTrace();
						System.out
								.println("Client: Error discovering remote data socket.");
					}
				}

				// The discovery is asynchronous as we do not know
				// how long is going to take
				try { // sleep as much as we want. Yes we
					// should implement asynchronous listener pipe...
					Thread.sleep(2000);
				} catch (Exception e) {
				}
			} catch (IOException e) {
				// found nothing! move on
			}
			System.out.print(".");
		}
		System.out.println();
	}

	public void receiveFile(String fileName) {

		System.out
				.println("listening for incoming binary stream to write it to file "
						+ fileName);

		File newFile = new File(fileName);
		FileOutputStream output = null;
		try {
			output = new FileOutputStream(newFile);
		} catch (FileNotFoundException e1) {
			System.err.println("could not open file to write: " + fileName);
			return;
		}

		try {
			Socket socket = dataInSocket.accept();
			InputStream in = socket.getInputStream();

			byte[] outBuffer = new byte[socket.getReceiveBufferSize()];
			int bytesReceived = 0;
			while ((bytesReceived = in.read(outBuffer)) > 0) {
				output.write(outBuffer, 0, bytesReceived);
			}
			socket.close();
			output.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void sendFile(String fileName) {
		System.out.println("Sending file " + fileName);
		System.out.println("Establishing data connection");

		JxtaSocket socket = null;
		try {
			socket = new JxtaSocket(netPeerGroup, dataOutPipeAdv,
			// timeout 10 seconds
					10000);
		} catch (IOException e) {
			System.err.println("timed out.");
		}

		File newFile = new File(fileName);
		FileInputStream input;
		try {
			input = new FileInputStream(newFile);
		} catch (FileNotFoundException e) {
			System.err.println("file not found: " + fileName);
			return;
		}

		try {
			OutputStream out = socket.getOutputStream();

			byte[] nextBytes = new byte[socket.getSendBufferSize()];
			int bytesRead = 0;
			while ((bytesRead = input.read(nextBytes)) > 0) {
				out.write(nextBytes, 0, bytesRead);
				System.out.println(".");
			}
			System.out.println("finished");

			out.flush();
			input.close();
			socket.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * Starts jxta
	 */
	public void startJxta() {
		try {
			// create, and Start the default jxta NetPeerGroup
			netPeerGroup = PeerGroupFactory.newNetPeerGroup();
		} catch (PeerGroupException e) {
			// could not instantiate the group, print the stack and exit
			System.out.println("fatal error : group creation failure");
			e.printStackTrace();
			System.exit(1);
		}
		if (6 < DEBUGLEVEL)
			System.out.println("Getting DiscoveryService");
		discovery = netPeerGroup.getDiscoveryService();
		if (6 < DEBUGLEVEL)
			System.out.println("Getting PipeService");
		pipes = netPeerGroup.getPipeService();

		/**
		 * TODO: clean local cache from Advertisements of last runtimes
		 */
		try {
			flushLocalCache();
		} catch (IOException ioe) {
			System.err
					.println("flushing local cache of advertisements was not successful. "
							+ "Please do it manually (remove directory /.jxta/cm)");
		}

		prepareDataSocket();
		prepareMsgPipe();
	}

	/**
	 * on startup the local cache is cluttered with old entries from the last
	 * run which are not up to date any longer because the other peers may have
	 * left the network or shut down. therefore this method offers the cleanout
	 * of the local cache.
	 * 
	 * @throws IOException
	 */
	public void flushLocalCache() throws IOException {
		Enumeration myLocalEnum = null;
		myLocalEnum = discovery.getLocalAdvertisements(DiscoveryService.ADV,
				"Name", "JXTASPEC:" + msgPipeName);
		if ((myLocalEnum != null) && myLocalEnum.hasMoreElements()) {
			LOG.log(Level.INFO,
					"Flush local cache : flushing advertisements... ");
			Advertisement adv = null;
			while (myLocalEnum.hasMoreElements()) {
				adv = (Advertisement) myLocalEnum.nextElement();
				discovery.flushAdvertisement(adv);
			}
		}

		myLocalEnum = discovery.getLocalAdvertisements(DiscoveryService.ADV,
				"Name", "JXTASPEC:" + dataSocketName);
		if ((myLocalEnum != null) && myLocalEnum.hasMoreElements()) {
			LOG.log(Level.INFO,
					"Flush local cache : flushing advertisements... ");
			Advertisement adv = null;
			while (myLocalEnum.hasMoreElements()) {
				adv = (Advertisement) myLocalEnum.nextElement();
				discovery.flushAdvertisement(adv);
			}
		}

	}

}