joineddownload.java

一个基于NetBeans平台开发的

					File riverFile = new File("riverfile/" + s + ".river");
                                        FileOutputStream fos = new FileOutputStream(riverFile);
                                        fos.write(fileData);
                                        fos.flush();
                                        fos.close();
                                        
					taskInfo.initJoinedProperties(s);
					this.filePieceLength = Integer.valueOf(taskInfo.getTaskPieceLength() ).intValue();
					
					joinedDownloadMessageBuffer.remove(inmsg);
//					meeageBufferIndex = 0;
					//将第0块设置为true，表示已经下载了"元文件"。
					taskInfo.setTaskFinishedPieceIndex("0");
                                        taskInfo.taskUpdata();
System.out.println("join 收到了块 0");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "join 收到了块 0。";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
					continue;
				}
				//如果返回的是具体的文件信息
				else {
                                    boolean[] finishedPs = taskInfo.getTaskFinishedPieces();
                                    int cameBlock = Integer.valueOf(block).intValue();
                                    if ( finishedPs[cameBlock] ){
                                        joinedDownloadMessageBuffer.remove(inmsg);
System.out.println("join 放弃了一个块");
                                        continue;
                                    }
                                    
                                    //设置文件块长度，如果是再次启动，这个值将是0，因为没有赋值，原来的情况是：只是在第一次启动时(请求0块后,133行附近)赋值一次。
                                    this.filePieceLength = Integer.valueOf(taskInfo.getTaskPieceLength() ).intValue();
					//一下的"文件"类，是为了向指定文件中写得到的数据。
					File file = null;
					BufferedRandomAccessFile raf = null;
					
					int filePiece = Integer.valueOf(block).intValue();
					--filePiece;					//实际要请求的文件索引

					if(taskInfo.getTaskRiverFileType().equals("single")){
						file = new File(taskInfo.getTaskFileName() + ".rs");
						raf = new BufferedRandomAccessFile(file,"rw");
						raf.seek((long)(filePiece*filePieceLength));
						raf.write(fileData);
					}else{
						String temp = taskInfo.getFileNameOfMultipleFromPiece(block);
						file = new File(temp + ".rs");
						raf = new BufferedRandomAccessFile(file,"rw");
						int pieceindex = Integer.valueOf(taskInfo.getFileDataIndex()).intValue();
						raf.seek( (long)(pieceindex * filePieceLength) );
						raf.write(fileData);
					}
                                raf.close();
System.out.println("join 收到了 块" + (filePiece+1) );
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "join 收到了 块" + (filePiece+1) + "。";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                                }
				
				//设置已经下载完的文件块
				taskInfo.setTaskFinishedPieceIndex(block);
				
				//设置已经下载完的文件长度
				int fileLength = Integer.valueOf(taskInfo.getTaskFinishedLength()).intValue();
				fileLength += (filePieceLength>fileData.length)?fileData.length:filePieceLength  ;
				taskInfo.setTaskFinishedLength(String.valueOf(fileLength));

				//设置完成百分比
				int totalLength = Integer.valueOf(taskInfo.getTaskFileLength()).intValue();
				float p1 = ((float)fileLength)/((float)totalLength);
				int p2 = Math.round(p1 * 1000);
				float p = p2 / (float)10;
				taskInfo.setTaskPercentage(String.valueOf(p) + "%");
				
				//从队列中删除处理后的Message对象
				joinedDownloadMessageBuffer.remove(inmsg);
                                taskInfo.taskUpdata();

        		}catch (IOException ioe ) {ioe.printStackTrace();}
			//解决CPU占用率过高的问题。
			try{Thread.sleep(5);} catch (Exception e) {}
		}
System.out.println("download 任务停止！");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "download 任务停止！";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
	}
	
        //参数 b 是请求的块数，"0"为元文件，
        //而第一个文件数据的索引号是"1"
	private boolean sendRequest(String b,int joinedRequestBuffer_index){
		OutputPipe outputPipe = null;
                PipeAdvertisement pipeAdv = null;
                
System.out.println("join 开始请求第"+ b +"块");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "join 开始请求第"+ b +"块";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
		while (true){
			try{
				pipeAdv = (PipeAdvertisement)joinedRequestBuffer.get(joinedRequestBuffer_index);
				outputPipe = ((PipeService)taskInfo.getTaskPipes()).createOutputPipe(pipeAdv,5000);
				
				/**如果这个消息的管道没有连接上，
				  *退出请求
                                 *向下一个joinedRequestBuffer 中 的通告进行发送请求
                                 *
				  */
				if (outputPipe == null)	{
                                    break;
				}
System.out.println("join 与一个peer建立了连接: ");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "join 与一个peer建立了连接.";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                                //封装数据
                                outmsg = new Message();
				//第一个数据信息，
				//管道通告。
				InputStreamMessageElement outMsgPipeElement = new InputStreamMessageElement("PipeAdv",mimetype,taskInfo.getTaskPipeadv().getDocument(mimetype).getStream(),null);
				outmsg.addMessageElement(outMsgPipeElement);
				
				//第二个数据信息，
				//类型
				ByteArrayMessageElement outMsgTypeElement = new ByteArrayMessageElement("Type",MimeMediaType.XMLUTF8,"Req".getBytes(),null);
				outmsg.addMessageElement(outMsgTypeElement);
				
				//第三个数据信息，
				//块数
				ByteArrayMessageElement outMsgBlockElement = new ByteArrayMessageElement("Block",MimeMediaType.XMLUTF8,b.getBytes(),null);
				outmsg.addMessageElement(outMsgBlockElement);
				
				//第四个数据信息，
				//具体的数据。
				ByteArrayMessageElement outMsgDataElement = new ByteArrayMessageElement("Data",MimeMediaType.XMLUTF8,"?".getBytes(),null);
				outmsg.addMessageElement(outMsgDataElement);
 
				boolean d = outputPipe.send(outmsg);

				outputPipe.close();                      //最后要察看是否可以关闭它//////////////

				if (d) {System.out.println("joined的Download任务成功发送了第"+b+"块的请求。");break;}
			}catch (IOException ioe ) {
System.out.println("一个 Peer 没有被连接，放弃与它的连接");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "一个 Peer 没有被连接，放弃与它的连接.";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                            joinedRequestBuffer.remove(pipeAdv);
                            return false;
                        }
		}
        return true;
	}
	
	//产生一个随机的请求文件块索引
	private String getRandomDataIndex(){
		boolean[] finishedPieces = taskInfo.getTaskFinishedPieces();
		int pieces = ((Integer)taskInfo.getTaskDictionary().get("pieces")).intValue()/20;
		Date date = new Date();
		Random r = new Random(date.getTime());
		int currentIndex = 0;

			while ( true ){
				currentIndex = r.nextInt(pieces);
                                ++currentIndex ;            //需要加一，因为随机数是从"0"开始，到随机数-1的范围，而"0"是元文件，所以第0块文件数据是"1(0+1)"
				if ( !finishedPieces[currentIndex] ) break;
			}

		return String.valueOf(currentIndex);
	}
        //下载到 100.0% 时的操作。
        private void do100(){
            if(taskInfo.getTaskRiverFileType().equals("single")){
                new File(taskInfo.getTaskFileName() + ".rs").renameTo(new File(taskInfo.getTaskFileName()));
                taskInfo.taskUpdata();
            }
            else{
                Hashtable d = (Hashtable)taskInfo.getTaskDictionary();
                int  size = ((Integer)d.get("path_index")).intValue();
                String rootPath = taskInfo.getTaskFileName();
                for ( int i=1;i<=size;i++ ){
                    String childPath = (String)d.get("path" + i);
                    File fileN = new File(rootPath + "/" + childPath);
                    if ( fileN.exists() ) continue;
                    else (new File(rootPath + "/" + childPath + ".rs")).renameTo(fileN);
                }
            }
        }
}