publishupload.java

来自「一个基于NetBeans平台开发的」· Java 代码 · 共 232 行

package com.sinpool.rivercrescent.Task;

import net.jxta.endpoint.*;
import net.jxta.pipe.*;
import net.jxta.document.*;
import net.jxta.protocol.*;

import java.io.*;
import java.util.Date;

import com.sinpool.rivercrescent.*;
import com.sinpool.rivercrescent.myutil.MyUtil;
import com.sinpool.rivercrescent.io.*;
import com.sinpool.rivercrescent.BufferZone.*;

public class PublishUpload extends Thread{
	private MessageBuffer messageBuffer = null;
	private PublishManagerControl publishManagerControl = null;
	private TaskInfo taskInfo = null;
        
	private MimeMediaType mimetype = new MimeMediaType("text","xml");
	private Message inmsg = new Message();
	private Message outmsg = null;
	private InputStream is = null;
	private PipeAdvertisement pipeAdv = null;
	private MessageElement me = null;
	private OutputPipe outputPipe = null;
	private int meeageBufferIndex = 0;
	private String block = null;
	//多文件时，文件块的长度，即pl
	private int filePieceLength = 0;
        //实际从文件中读出了多少字节数。
        private int readFileDataLength = 0;
        
        private String[] thisStatus = new String[2];
        
	public PublishUpload(MessageBuffer messageBuffer,TaskInfo ti,PublishManagerControl publishManagerControl){
                this.messageBuffer = messageBuffer;
		this.taskInfo = ti;
		this.publishManagerControl = publishManagerControl;
                this.filePieceLength = Integer.valueOf(taskInfo.getTaskPieceLength()).intValue();
	}
	
	public void run(){
		long startTime = 0,endTime = 0;
		Date date = new Date();
		byte[] fileData;			//根据消息请求，从现有文件读取的、用于返回的数据
System.out.println("任务开始！");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "上传任务开始！";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);

		while ( publishManagerControl.searchUploadManagerState() ){
			try{
                                BufferedReader br ;
				inmsg = (Message)messageBuffer.get(meeageBufferIndex++);
				
				me = inmsg.getMessageElement("Type");
				is = me.getStream();
				br = new BufferedReader(new InputStreamReader(is) );
				String messageType = br.readLine();
				
				//确保从缓冲区取出的Message，一定是"请求"
				if ( messageType.equals("Res") ) {
                                    messageBuffer.remove(inmsg);
                                    meeageBufferIndex = 0;
System.out.println("publish 放弃了一个消息");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "publish 放弃了一个消息！";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                                    continue;
                                }
                                
				me = inmsg.getMessageElement("PipeAdv");
				is = me.getStream();
				pipeAdv = (PipeAdvertisement)AdvertisementFactory.newAdvertisement(mimetype,is);
				outputPipe = ((PipeService)taskInfo.getTaskPipes()).createOutputPipe(pipeAdv,5000);
				
				/**如果这个消息的管道没有连接上，
				  *
				  *如果索引值等于消息队列的大小时，
				  *meeageBufferIndex索引回0，
				  *从头读取消息。
				  *
				  */
				if (outputPipe == null)	{
					if (meeageBufferIndex == messageBuffer.size() )
						meeageBufferIndex = 0;
					continue;
				}
				
				me = inmsg.getMessageElement("Block");
				is = me.getStream();
                                br = new BufferedReader(new InputStreamReader(is) );
				block = br.readLine();
                                
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "收到了第" +  block + "块的请求。";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);

				//处理文件数据。
				try{
					fileData = this.getFileData(block);
				}catch (IOException ioe) {
                                    System.out.println("没能打开源文件！");
                                    continue;
                                }
				
				//制作新的输出消息
                                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,"Res".getBytes(),null);
				outmsg.addMessageElement(outMsgTypeElement);
				
				//第三个数据信息，
				//块数
				ByteArrayMessageElement outMsgBlockElement = new ByteArrayMessageElement("Block",MimeMediaType.XMLUTF8,block.getBytes(),null);
				outmsg.addMessageElement(outMsgBlockElement);
				
				//第四个数据信息，
				//具体的数据。
				ByteArrayInputStream dataPiece = new ByteArrayInputStream(fileData);
				InputStreamMessageElement outMsgDataElement = new InputStreamMessageElement("Data",mimetype,dataPiece,null);
				outmsg.addMessageElement(outMsgDataElement);
				
				startTime = date.getTime();
				boolean b = outputPipe.send(outmsg);
				if (b) {
					endTime = date.getTime();
					int time = (int)((endTime - startTime)/1000);
					if (time == 0){
                                            if ( readFileDataLength < filePieceLength )
                                                taskInfo.setTaskUploadSpeed(String.valueOf( readFileDataLength ) );
                                            else
						taskInfo.setTaskUploadSpeed(String.valueOf(filePieceLength));
                                        }
					else
						taskInfo.setTaskUploadSpeed(String.valueOf(filePieceLength/(int)(time)));
				}
				messageBuffer.remove(inmsg);
				meeageBufferIndex = 0;
				outputPipe.close();				//最后要察看是否可以关闭它//////////////
                                taskInfo.taskUpdata();
			}catch (IOException ioe ) {
System.out.println("一个 Peer 没有被连接，放弃与它的连接");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "一个 Peer 没有被连接，放弃与它的连接！";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                            messageBuffer.remove(inmsg);
                            meeageBufferIndex = 0;
                            continue;
                        }
System.out.println("publish任务上传了第"+block+"块的响应.");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "publish任务上传了第"+block+"块的响应.";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
			try{Thread.sleep(5);} catch (Exception e) {}
		}
System.out.println("任务停止！");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "publish 任务停止！";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
	}
	/*b是"0"，说明请求元文件；
	 *b从"1"开始，说明请求源文件；
	 *而在文件的索引中以"0"开始，
	 *所以要"--filePiece"，
	 *使得"1"变成文件的所以"0"，
	 *而"0-1 = -1"，当filePiece为"-1"时，
	 *说明请求元文件。
	 *
	 */
	private byte[] getFileData(String b) throws IOException{
		File file = null;
		BufferedRandomAccessFile raf = null;
		FileInputStream fis = null;
		
		int filePiece = Integer.valueOf(b).intValue();
		--filePiece;
		byte[] data = null;

		//请求元文件
		if (filePiece == -1){
			file = new File((String)taskInfo.getTaskRiverscentFileName());

			fis = new FileInputStream(file);
			data = new byte[(int)file.length()];
                        fis.read(data);
                        
                        fis.close();
		}
		//请求数据文件
		//需要判断是单文件情况还是多文件情况
		else {
                    if(taskInfo.getTaskRiverFileType().equals("single")){
				file = new File(taskInfo.getTaskFileName());
                                
				raf = new BufferedRandomAccessFile(file,"r");
				data = new byte[filePieceLength];
				raf.seek((long)(filePiece*filePieceLength));
				readFileDataLength = raf.read(data);

                                raf.close();
			}else{
				String temp = taskInfo.getFileNameOfMultipleFromPiece(b);
				file = new File(temp);
				raf = new BufferedRandomAccessFile(file,"r");
				data = new byte[filePieceLength];
				int pieceindex = Integer.valueOf(taskInfo.getFileDataIndex()).intValue();
				raf.seek( (long)(pieceindex * filePieceLength) );
				readFileDataLength = raf.read(data);
                                
                                raf.close();
			}
                
                    //如果读到了最后一个不足"文件块"长度的字节数组，使用这个方式返回。
                    if ( readFileDataLength < filePieceLength ){
                        byte[] lastData = new byte[readFileDataLength];
                        for ( int i=0;i<readFileDataLength;i++)
                            lastData[i] = data[i];
                        return lastData;
                    }
                }
		return data;
	}
}