messagequeue.java

移动empp网关

package cn.madhouse.gateway.empp.service;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;

import java.util.Map;
import java.util.HashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

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

public class MessageQueue implements Serializable {
	
	private String fileName;
	
	private ConcurrentLinkedQueue<DeliverRequest> moRequestQueue;
	private ConcurrentLinkedQueue<DeliverResponse> moResponseQueue;
	private ConcurrentLinkedQueue<SubmitRequest> mtRequestQueue;
	private ConcurrentLinkedQueue<SubmitResponse> mtResponseQueue;
	private ConcurrentLinkedQueue<DeliverReport> reportQueue;
	private ConcurrentLinkedQueue<byte[]> otherQueue;
	
	private Map<Integer,Long> sequenseId_idMap;
	
	public void setFileName(String fileName) {
		this.fileName = fileName;
	}
	public String getFileName() {
		return fileName;
	}
	public ConcurrentLinkedQueue<DeliverRequest> getMoRequestQueue() {
		return moRequestQueue;
	}
	public ConcurrentLinkedQueue<DeliverResponse> getMoResponseQueue() {
		return moResponseQueue;
	}
	public ConcurrentLinkedQueue<SubmitRequest> getMtRequestQueue() {
		return mtRequestQueue;
	}
	public ConcurrentLinkedQueue<SubmitResponse> getMtResponseQueue() {
		return mtResponseQueue;
	}
	public ConcurrentLinkedQueue<byte[]> getOtherQueue() {
		return otherQueue;
	}
	public ConcurrentLinkedQueue<DeliverReport> getReportQueue() {
		return reportQueue;
	}
	public Map<Integer,Long> getSequenseId_idMap() {
		return sequenseId_idMap;
	}
	
	public void init() throws FileNotFoundException, IOException, ClassNotFoundException {
		MessageQueue msgQueue = readSerial(new File(fileName));
		if(msgQueue != null) {
			moRequestQueue = msgQueue.getMoRequestQueue();
			moResponseQueue = msgQueue.getMoResponseQueue();
			mtRequestQueue = msgQueue.getMtRequestQueue();
			mtResponseQueue = msgQueue.getMtResponseQueue();
			reportQueue = msgQueue.getReportQueue();
			otherQueue = msgQueue.getOtherQueue();
		} else {
			moRequestQueue = new ConcurrentLinkedQueue<DeliverRequest>();
			moResponseQueue = new ConcurrentLinkedQueue<DeliverResponse>();
			mtRequestQueue = new ConcurrentLinkedQueue<SubmitRequest>();
			mtResponseQueue = new ConcurrentLinkedQueue<SubmitResponse>();
			reportQueue = new ConcurrentLinkedQueue<DeliverReport>();
			otherQueue = new ConcurrentLinkedQueue<byte[]>();
		}
		sequenseId_idMap = new HashMap<Integer,Long>();
	}
	
	public void persist() throws IOException {
		MessageQueue.writeSerial(new File(fileName), this);
	}
	
	public static void writeSerial(File file, MessageQueue msgQueue) throws IOException {
		if(!file.exists()) file.mkdir();
		FileOutputStream outputFile = new FileOutputStream(file);
        ObjectOutputStream serializeStream = new ObjectOutputStream(outputFile);
        serializeStream.writeObject(msgQueue);
        serializeStream.flush();
        serializeStream.close();
        outputFile.close();
	}
	
	public static MessageQueue readSerial(File file) throws FileNotFoundException, IOException, ClassNotFoundException {
        if (!file.exists()) {
        	return null;
        } else {
        	 FileInputStream inputFile = new FileInputStream(file);
             ObjectInputStream serializedStream = new ObjectInputStream(inputFile);
             MessageQueue msgQueue = (MessageQueue) serializedStream.readObject();
             serializedStream.close();
             inputFile.close();
             file.delete();
             return msgQueue;
        }
	}
	
}