sendoutthread.java

移动empp网关

package cn.madhouse.gateway.empp.service;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

import cn.madhouse.gateway.empp.msgformat.*;

public class SendOutThread extends Thread {
	
	protected int sleepTime;
	protected MessageQueue msgQueue;
	protected Map<String,EmppSocket> emppSocketMap;
	
	public SendOutThread(int sleepTime, MessageQueue msgQueue, Map<String,EmppSocket> emppSocketMap) {
		this.sleepTime = sleepTime;
		this.msgQueue = msgQueue;
		this.emppSocketMap = emppSocketMap;
	}
	
	@Override
	public void run() {
		ConcurrentLinkedQueue<SubmitRequest> mtRequestQueue = msgQueue.getMtRequestQueue();
		ConcurrentLinkedQueue<DeliverResponse> moResponseQueue = msgQueue.getMoResponseQueue();
		
		while(true) {
			//System.out.println("sendout");
			SubmitRequest mtReq;
			DeliverResponse moResp;
			do {
				mtReq = mtRequestQueue.poll();
				if(mtReq != null) {
					EmppSocket emppSocket = (EmppSocket) emppSocketMap.get(mtReq.getServiceId());
					emppSocket.write(mtReq.toByte());
					//System.out.println("mtReq commandId=" + mtReq.getCommandId());
					sleepT(sleepTime);
				} else break;
			} while (true);
			
			do {
				moResp = moResponseQueue.poll();
				if(moResp != null) {
					EmppSocket emppSocket = (EmppSocket) emppSocketMap.get(moResp.getServiceId());
					emppSocket.write(moResp.toByte());
					//System.out.println("moResp commandId=" + moResp.getCommandId());
					sleepT(sleepTime);
				} else break;
			} while (true);
			
			sleepT(sleepTime);
		}
	}
	
	private void sleepT(int million) {
		try {
			sleep(million);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}