rfidreadereventpush.java

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

// Copyright (c) 2000 Just Objects B.V. <[e-mail]>
// Distributable under LGPL license. See terms of license at gnu.org.

package hfut.wispy.server;

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

import hfut.wispy.ArrayBuffer;
import hfut.wispy.Buffer;
import hfut.wispy.ListBuffer;
import hfut.wispy.NoBuffer;
import hfut.wispy.RFIDReader;
import hfut.wispy.TagDataPacket;
import hfut.wispy.webservice.GlobalSettings;
import hfut.wispy.webservice.RFIDReaderSettings;
import hfut.wispy.webservice.RFIDReaderSettings.ReaderModel;

import java.util.Date;
import java.util.UUID;

/**
 * Event sources that push events (for testing).
 * 
 * @author Just van den Broecke - Just Objects &copy;
 * @version $Id: TestEventPushSources.java,v 1.10 2007/11/09 13:16:57 justb Exp $
 */
public class RFIDReaderEventPush {

	/**
	 * Produces events from REAL stocks from the AEX.
	 */
	static public class RFIDReaderEventPushSource1 implements EventSource,
			Runnable {
		Thread thread = null;

		boolean active = true;

		public static Date lastUpdateDate = new Date();

		private int restarts = 1;

		static RFIDReader reader = new RFIDReader();

		static Buffer buffer = ListBuffer.getInstance();

		static GlobalSettings globalSettings = GlobalSettings.getInstance();

		static RFIDReaderSettings readerSettings = new RFIDReaderSettings(
				ReaderModel.YWG804_6c_cycle);

		static int PushTimes = 1;

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

		/**
		 * Activate the event source.
		 */
		synchronized 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");
		}

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

		/**
		 * Deactivate the event source.
		 */
		synchronized public void passivate() {
			e("passivating...");
			active = false;
			stopThread();
			// Mark the cache modified so we'll send the contents
			// on the next activation.
			for (int i = 0; i < buffer.getBufferLength(); i++) {
				buffer.getTagPacket(i).modified = true;
			}
			e("passivated");
		}

		/**
		 * Deactivate the event source.
		 */
		synchronized public void stop() {
		}

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

			initCache();
			updateCache();
			while (active) {
				this.thread.setPriority(Thread.MIN_PRIORITY);
				while (globalSettings.IsStartReader) {
					this.thread.setPriority(Thread.NORM_PRIORITY);
					if (count % 2 == 0) {
						updateCache();
						Date now = new Date();
						p("距上次updateCache()时间 : "
								+ (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(1, 20));
					} catch (InterruptedException ie) {
						ie.printStackTrace();
						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 < buffer.getBufferLength(); i++) {
				TagDataPacket nextTagPacket = buffer.getTagPacket(i);
				// Publish modified stocks
				if (nextTagPacket.modified && nextTagPacket.SuccessRead) {
					publishStock(i, nextTagPacket.AntennaNo,
							nextTagPacket.DevieceNo, nextTagPacket.hasMoreTag,
							nextTagPacket.HaveReadTimes, nextTagPacket.id,
							nextTagPacket.FirstReadTime,
							nextTagPacket.LastReadTime, nextTagPacket.ReadTime,
							nextTagPacket.TagData, nextTagPacket.UnReadTimes);
					nextTagPacket.modified = false;
					// wispy no sleep test
					try {
						Thread.sleep(200);
					} catch (InterruptedException ie) {
						return;
					}
				}
				// test---不管modified了，全都push一把
				// publishStock(i, nextStock.TagData, nextStock.HaveReadTimes,
				// nextStock.ReadTime);
			}
		}

		private void publishStock(int index, byte AntennaNo, int DevieceNo,
				boolean hasMoreTag, int HaveReadTimes, UUID id, Date FirstReadTime,
				Date LastReadTime, Date ReadTime, byte[] TagData,
				int UnReadTimes) {
			Event event = Event.createDataEvent("/wispy/RFIDReader");
			event.setField("PushTimes", PushTimes);
			event.setField("AntennaNo", Byte.toString(AntennaNo));
			event.setField("DevieceNo", DevieceNo + "");
			event.setField("hasMoreTag", Boolean.toString(hasMoreTag));
			event.setField("HaveReadTimes", HaveReadTimes + "");
			event.setField("id", id.toString());
			event.setField("FirstReadTime", FirstReadTime.toString());
			event.setField("LastReadTime", LastReadTime.toString());
			event.setField("ReadTime", ReadTime.toString());
			event.setField("TagData", RFIDReader.Bytes2HexString(TagData));
			event.setField("UnReadTimes", UnReadTimes);
			PushTimes++;
			Dispatcher.getInstance().multicast(event);
		}

		private void publishStock(int index, byte[] tagData, int HaveReadTimes,
				Date ReadTime) {
			byte[] temp = new byte[12];
			if (tagData != null)
				temp = tagData;
			Event event = Event.createDataEvent("/wispy/RFIDReader");
			event.setField("number", index + "");
			event.setField("name", RFIDReader.Bytes2HexString(temp));
			event.setField("rate", HaveReadTimes + "");
			event.setField("time", ReadTime.toString());
			Dispatcher.getInstance().multicast(event);
		}

		private void publishStock(int index, String name, String rate,
				String time) {
			Event event = Event.createDataEvent("/wispy/RFIDReader");
			event.setField("number", index + "");
			event.setField("name", name);
			event.setField("rate", rate);
			event.setField("time", time);
			// p("publish: nr=" + index + " name=" + name + " rate=" + rate);
			// p("[sending updates] --rate-- : " + rate);
			// p("[sending updates] --time-- : " + time);
			Dispatcher.getInstance().multicast(event);
		}

		/**
		 * 私有。被passivate()调用
		 */
		private void stopThread() {
			if (thread != null) {
				thread.interrupt();
				thread = null;
			}
		}

		private static void initCache() {
			reader = new RFIDReader();
			if (!reader.GetOpenState())
				reader.Open();
		}

		public synchronized static void updateCache() {
			reader.Identify(readerSettings.Antenna, readerSettings.IsCycleRead);
			for (int i = 0; i < readerSettings.CycleReadTime; i++) {
				TagDataPacket tag = buffer.MyAdd(reader.Report());
				if (tag != null) {
					p(tag.getFullStr());
				}
				buffer.refresh();
				p("now listBuffer length = " + buffer.getBufferLength());
			}
			reader.PowerOff();
		}

		public boolean isActive() {
			return active;
		}

		public void setActive(boolean active) {
			this.active = active;
		}
	}

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

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

}

/*
 * $Log: TestEventPushSources.java,v $ Revision 1.10 2007/11/09 13:16:57 justb
 * use Rand from util package (and and Rand.java to pushlet client jar
 * 
 * Revision 1.9 2005/02/28 09:14:56 justb sessmgr/dispatcher factory/singleton
 * support
 * 
 * Revision 1.8 2005/02/21 16:59:17 justb SessionManager and session lease
 * introduced
 * 
 * Revision 1.7 2005/02/18 10:07:23 justb many renamings of classes (make names
 * compact)
 * 
 * Revision 1.6 2005/02/18 09:54:15 justb refactor: rename Publisher Dispatcher
 * and single Subscriber class
 * 
 * Revision 1.5 2004/09/03 22:35:37 justb Almost complete rewrite, just checking
 * in now
 * 
 * Revision 1.4 2004/03/10 15:45:55 justb many cosmetic changes
 * 
 * Revision 1.3 2003/08/15 08:37:41 justb fix/add Copyright+LGPL file headers
 * and footers
 * 
 * Revision 1.2 2003/05/18 16:15:08 justb support for XML encoded Events
 * 
 * Revision 1.1.1.1 2002/09/24 21:02:33 justb import to sourceforge
 * 
 * Revision 1.1.1.1 2002/09/20 22:48:20 justb import to SF
 * 
 * Revision 1.1.1.1 2002/09/20 14:19:02 justb first import into SF
 * 
 * Revision 1.6 2001/02/18 23:45:13 just fixes for AEX
 * 
 * Revision 1.5 2000/10/30 14:16:09 just no message
 * 
 * Revision 1.4 2000/09/24 21:02:43 just chnages due to changed webpage in
 * debeurs.nl
 * 
 * Revision 1.3 2000/08/31 12:49:50 just added CVS comment tags for log and
 * copyright
 * 
 * 
 */