testeventpushsource2.java

Java下Comet的实现框架Pushlet例子。 实现实时推送数据到客户端。 服务器每隔30-500MS产生一个随机数

package nl.justobjects.pushlet.test;

import hfut.wispy.RFIDReader;
import hfut.wispy.TagDataPacket;
import hfut.wispy.test.ReportThread;
import hfut.wispy.test.TagDataPackets;

import java.util.Date;
import java.util.Vector;

import nl.justobjects.pushlet.core.Dispatcher;
import nl.justobjects.pushlet.core.Event;
import nl.justobjects.pushlet.core.EventSource;
import nl.justobjects.pushlet.util.Rand;

public class TestEventPushSource2 {

	static public class RFIDReaderEventPush implements EventSource, Runnable {

		Thread thread = null;

		volatile boolean active = false;

		// Since Baan has been thrown out...
		public final static int NR_OF_STOCKS = 5;

		static int a = 0;

		public static Date lastUpdateDate = new Date();

		public final static String EMPTY = "wait...";

		private int restarts = 1;

		public volatile static Vector<TagDataPacket> tagsCache = null;

		public RFIDReaderEventPush() {
			TagDataPackets.initData();
			tagsCache = TagDataPackets.getStocksCache();
		}

		public void run() {
			p("run()");
			// Publish cache content (if any) first.
			publishStocks();
			int count = 2; // enforce update first

			updateCache();
			while (active) {

				// Only do work if active
				// Update cache every 10 secs.
				// if (count % 2 == 0) {
				// updateCache();
				// Date now = new Date();
				// p("lastUpdateDate : "
				// + (now.getTime() - lastUpdateDate.getTime() + "毫秒"));
				// lastUpdateDate = now;
				// }
				count++;

				// Do updates for changed stock rates
				publishStocks();

				// If we were interrupted just return.
				if (thread == null || thread.isInterrupted()) {
					break;
				}

				// Sleep 2 secs before sending next updates
				try {
					Thread.sleep(Rand.randomLong(300, 3000));
				} catch (InterruptedException ie) {
					break;
				}
			}

			// Loop terminated: reset vars
			thread = null;
			active = false;

		}

		private void publishStocks() {
			// p("sending updates");
			// Publish only modified stocks from the cache
			for (int i = 0; i < NR_OF_STOCKS; i++) {
				TagDataPacket nextTag = (TagDataPacket) tagsCache.elementAt(i);
				// Publish modified stocks
				if (nextTag.modified) {
					publishStock(i, nextTag.TagData, nextTag.HaveReadTimes,
							nextTag.ReadTime);
					nextTag.modified = false;
					// wispy no sleep test
					try {
						Thread.sleep(200);
					} catch (InterruptedException ie) {
						return;
					}
				}
			}
		}

		public void publishStock(int index, byte[] tagData, int HaveReadTimes,
				Date ReadTime) {
			Event event = Event.createDataEvent("/stocks/aexonlineRFIDReader");
			event.setField("number", index + "");
			event.setField("tagData", RFIDReader.Bytes2HexString(tagData));
			event.setField("HaveReadTimes", HaveReadTimes + "");
			event.setField("ReadTime", ReadTime.toString());
			// p("publish: nr=" + index + " name=" + name + " rate=" + rate);
			// p("[sending updates] --rate-- : " + rate);
			// p("[sending updates] --time-- : " + time);
			Dispatcher.getInstance().multicast(event);
		}

		public static void updateCache() {
			ReportThread r = new ReportThread();
			r.active = true;
			Thread thread = new Thread(r);
			thread.start();

		}

		public void activate() {
			e("activating...");
			// Stop a possibly running thread
			stopThread();

			// Start new thread and
			thread = new Thread(this, "RFIDReaderPublisher-" + (restarts++));
			active = true;
			thread.start();
			p("[---Thread---] : RFIDReaderPublisher-" + restarts);
			e("activated");
		}

		private void stopThread() {
			if (thread != null) {
				thread.interrupt();
				thread = null;
			}
		}

		public Thread getThread() {
			if (thread != null) {
				return thread;
			}
			return null;
		}

		public void passivate() {
			// TODO 自动生成方法存根

		}

		public void stop() {
			// TODO 自动生成方法存根

		}
	}

	/**
	 * Util: stderr print method.
	 */
	public static void e(String s) {
		System.out.println("AEXStocksEventPushSource: " + s);
	}

	/**
	 * Util: stdout print method.
	 */
	public static void p(String s) {
		System.out.println(s);
	}

}