jointasklistener.java

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

/**类似publish的任务监听器，<br>
 *专门负责向JoinedDownloadMessageBuffer和JoinedUploadMessageBuffer中<br>
 *添加缓冲数据。<br>
 *
 *与publish不同的是：<br>
 *此监听器负责两个缓冲区，<br>
 *分别给上传和下载两个独立的任务使用。<br>
 *
 *它还负责启动两个具体工作的任务对象：<br>
 *
 *其中上传的任务与publish没有什么区别，<br>
 *直接加缓冲区即可等待数据的到来；<br>
 *而，下载的任务有些特别：<br>
 *要通过一个方法显示的启动下载任务线程，<br>
 *因为下载任务的缓冲区只有发出请求后，<br>
 *才会有响应数据到达JoinedDownloadMessageBuffer。<br>
 *
 *而发送请求需要首先发现其他的管道通告，<br>
 *所以在打开下载任务前，<br>
 *首先构造一个JoinedDownloadFindRequestListener监听器，<br>
 *这个监听器是把搜索的管道通告存入joinedRequestBuffer，<br>
 *以便下载任务在发送请求时，<br>
 *从joinedRequestBuffer缓冲区取出同奥。<br>
 *
 *@author Sinpool.<br>
 *
 *@version 1.0<br>
 */

package com.sinpool.rivercrescent.Listener;

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

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

import com.sinpool.rivercrescent.Task.*;
import com.sinpool.rivercrescent.BufferZone.*;
import com.sinpool.rivercrescent.interfaces.*;
import com.sinpool.rivercrescent.myutil.MyUtil;

public class JoinTaskListener implements MyPipeMsgListener{
	private JoinedManagerControl joinedManagerControl = new JoinedManagerControl();
	private JoinedDownloadMessageBuffer jdm = new JoinedDownloadMessageBuffer();
	private JoinedUploadMessageBuffer jum = new JoinedUploadMessageBuffer();
	private JoinedRequestBuffer jrb = new JoinedRequestBuffer();
        private JoinedManagerControl jdfrlManagerControl = new JoinedManagerControl();

	private boolean flag = true;
	private TaskInfo taskInfo = null;
        private JoinedDownloadFindRequestListener jdfrl = null;
	
        private String[] thisStatus = new String[2];
	/**此方法是构造方法，<br>
	 *需要指出此次启动监听器是否为刚加入组而启动的。<br>
	 *
	 *@param ti 完成构造的此任务的TaskInfo对象；<br>
	 *@param first 如果是刚加入组后，第一次启动此监听器，设置为true，<br>
	 *				这个值可以通过JFrame初始化而得到是否原来存在这个任务。<br>
	 */
	public JoinTaskListener(TaskInfo ti){
		this.taskInfo = ti;
	}
	
	/**把收到的消息数据分类存入缓冲区，<br>
	 *Res类型的数据存入JoinedDownloadMessageBuffer
	 *Req类型的消息存入JoinedUploadMessageBuffer
	 */
	public void pipeMsgEvent(PipeMsgEvent event){
		Message msg = event.getMessage();
		BufferedReader br = null;
		InputStream is = null;
		String messageType = null;
		MessageElement me = null;

		try{
			me = msg.getMessageElement("Type");
			is = me.getStream();
			br = new BufferedReader(new InputStreamReader(is) );
			messageType = br.readLine();
		}
		catch (IOException ioe) {}
		
		if ( messageType.equals("Res")){
                        //如果已经下载结束，就不在接受响应了。
                        if ( taskInfo.getTaskPercentage().equals("100.0%") ) return;
			jdm.addElement(msg);
System.out.println("joinTaskListener 收到了一个 Res(响应) 消息。");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "joinTaskListener 收到了一个 Res(响应) 消息。";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                }
		else {
			jum.addElement(msg);
System.out.println("joinTaskListener 收到了一个 Req(请求) 消息。");
thisStatus[0] = MyUtil.getCurrentTime();
thisStatus[1] = "joinTaskListener 收到了一个 Req(请求) 消息。";
taskInfo.getTaskStatusTableModel().addRow(thisStatus);
                }
		
		if (flag){
			flag = false;
			JoinedUpload ju = new JoinedUpload(jum,taskInfo,joinedManagerControl);
			
			ju.start();
		}
	}
	
	/**Join任务的特殊之处，<br>
	 *需要使用此方法显示启动现在监听器。<br>
	 */
	public void startJoinDownloadListener(){
		jdfrl = new JoinedDownloadFindRequestListener(jrb,taskInfo,jdfrlManagerControl);
		taskInfo.getTaskPeerGroup().getDiscoveryService().addDiscoveryListener(jdfrl);
		jdfrl.start();
		
		//稍等一会儿，以便JoinedDownloadFindRequestListener监听器先发现管道通告。
		try { Thread.sleep(5000);} catch (Exception e){}
		JoinedDownload jd = new JoinedDownload(jdm,taskInfo,joinedManagerControl,jrb,jdfrl);
		jd.start();
	}
	
	/**关闭此任务监听器，<br>
	 *可以达到关闭此join任务的目的。<br>
	 */
	public void stopListener(){
		joinedManagerControl.stopJoinedManager();
                jdfrlManagerControl.stopJoinedManager();
                jdfrl.interrupt();
                
                //唤醒等待中的download 线程
                Message wakeup_d_msg = new Message();
                ByteArrayMessageElement wakeup_d_MsgTypeElement = new ByteArrayMessageElement("Type",MimeMediaType.XMLUTF8,"Req".getBytes(),null);
                wakeup_d_msg.addMessageElement(wakeup_d_MsgTypeElement);
                jdm.addElement(wakeup_d_msg);
                //唤醒等待中的upload 线程
                Message wakeup_u_msg = new Message();
                ByteArrayMessageElement wakeup_u_MsgTypeElement = new ByteArrayMessageElement("Type",MimeMediaType.XMLUTF8,"Res".getBytes(),null);
                wakeup_u_msg.addMessageElement(wakeup_u_MsgTypeElement);
                jum.addElement(wakeup_u_msg);
	}
}