testeventpushsources1.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 nl.justobjects.pushlet.test;

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.webservice.GlobalSettings;

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

/**
 * 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 TestEventPushSources1 {

	static public class AEXStocksEventPushSourceABN {
		String pageURL = "http://ri2.rois.com/E36msPtnZC0e15CVb4KT97JAGfGSfCcrvv6*FcyZIoNyh/CTIB/RI2APISNAP?RIC=0%23.AEX&FORMAT=XML";
		// This could be further expanded: getting the Reuters AEX stocks
		// as XML from ABN with this URL, but we may get into legal problems...
	}

	/**
	 * Produces events from REAL stocks from the AEX.
	 */
	static public class AEXStocksEventPushSource1 implements EventSource,
			Runnable {
		/**
		 * Here we get our stocks from.
		 */
		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<Stock> stocksCache = new Vector(
				NR_OF_STOCKS);

		static GlobalSettings globalSettings = GlobalSettings.getInstance();

		public AEXStocksEventPushSource1() {
			for (int i = 0; i < NR_OF_STOCKS; i++) {
				stocksCache.addElement(new Stock());
			}
			// updateCache();
		}

		/**
		 * Activate the event source.
		 */
		synchronized public void activate() {
			e("activating...");
			// Stop a possibly running thread
			stopThread();

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

		synchronized public Thread getThread() {
			if (thread != null) {
				return thread;
			}
			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 < NR_OF_STOCKS; i++) {
			// ((Stock) stocksCache.elementAt(i)).modified = true;
			// }

			e("passivated");
		}

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

		public void run() {
			p("run()");
			// Publish cache content (if any) first.
			publishStocks();

			int count = 2; // enforce update first

			while (active) {
				this.thread.setPriority(Thread.MIN_PRIORITY);
				if (globalSettings.IsStartReader) {
					this.thread.setPriority(Thread.NORM_PRIORITY);
					// 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(500, 1500));
					} catch (InterruptedException ie) {
						break;
					}
				} else {
					try {
						Thread.sleep(2000);
					} catch (InterruptedException e) {
						// TODO 自动生成 catch 块
						e.printStackTrace();
					}
				}
			}
			// 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++) {
				Stock nextStock = (Stock) stocksCache.elementAt(i);

				// Publish modified stocks
				if (nextStock.modified) {
					publishStock(i, nextStock.name, nextStock.rate,
							nextStock.time);
					nextStock.modified = false;
					// wispy no sleep test
					try {
						Thread.sleep(200);
					} catch (InterruptedException ie) {
						return;
					}
				}
			}
		}

		public void publishStock(int index, String name, String rate,
				String time) {
			Event event = Event.createDataEvent("/stocks/aexonlinewispy");
			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);
		}

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

		public static void updateCache(String name, String rate, String time) {
			e("");
			System.out.println("[updateCache]   ----" + "SWT" + "----");
			Stock currentStock1 = (Stock) stocksCache.elementAt(3);
			currentStock1.name = name;
			currentStock1.rate = rate;
			currentStock1.time = time;
			currentStock1.modified = true;
			e("");
		}

		public static void updateCache() {
			// p("updating Cache");

			// Get the line with all stocks from HTML page
			// String stocksLine = getStocksLine();
			// if ("".equals(stocksLine)) {
			// e("updateCache: stocksLine == null");

			if (true) {
				System.out.println("[updateCache]    ----" + a + "----");
				Stock currentStock1 = (Stock) stocksCache.elementAt(0);
				currentStock1.name = "tnt---111";
				currentStock1.rate = a + "";
				currentStock1.time = new Date().toString();
				a++;
				currentStock1.modified = true;

				System.out.println("[updateCache]   ----" + a + "----");
				Stock currentStock2 = (Stock) stocksCache.elementAt(1);
				currentStock2.name = "ccc---222";
				currentStock2.rate = a + "";
				currentStock2.time = new Date().toString();
				a++;
				currentStock2.modified = true;

				System.out.println("[updateCache]    ----" + a + "----");
				Stock currentStock3 = (Stock) stocksCache.elementAt(2);
				currentStock3.name = "wyy---333";
				currentStock3.rate = a + "";
				currentStock3.time = new Date().toString();
				a++;
				currentStock3.modified = true;

				return;
			}

			// Parse the stocksline and put in cache.
			// Beware: this is the messy part!!
			// We assume that stock/names and rates are located at
			// regular positions in the line.
			//			
			// String delim = "<>";
			// StringTokenizer st = new StringTokenizer(stocksLine, delim);
			// String nextToken = "";
			// int count = 0;
			// String nextStock = "";
			// String nextQuote = "";
			// String currentQuote = null;
			// int index = -1;
			// while (st.hasMoreTokens()) {
			// nextToken = st.nextToken();
			// count++;
			// // The <TD> with the stock name
			// if ((count - 5) % 57 == 0) {
			// p("c=" + count + " s=" + nextToken);
			// nextStock = nextToken;
			// }
			//
			// // The <TD> with the stock rate
			// if ((count - 10) % 57 == 0) {
			// nextQuote = nextToken;
			// index++;
			// p("c=" + count + " val=" + nextQuote);
			// Stock currentStock = (Stock) stocksCache.elementAt(index);
			//
			// // Only update new or modified stocks
			// if (EMPTY.equals(currentStock.rate) ||
			// !currentStock.rate.equals(nextQuote)) {
			// p("modified: " + nextStock);
			// currentStock.name = nextStock;
			// currentStock.rate = nextQuote;
			// currentStock.modified = true;
			// }
			// }
			// }
		}

		public boolean isActive() {
			return active;
		}

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

		public static Vector getStocksCache() {
			return stocksCache;
		}

		public static void setStocksCache(Vector stocksCache) {
			AEXStocksEventPushSource1.stocksCache = stocksCache;
		}
	}

	/**
	 * 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);
	}

	public static void main(String[] args) {
		// new TestEventPushSources$AEXStocksEventPushSource();
	}

	// private void sendUpdates() {
	// p("sending updates");
	// // In any case send a random stock value by
	// // making it modified, just to see something moving...
	// // int randomIndex = Rand.randomInt(0, NR_OF_STOCKS - 1);
	// // Stock randomStock = (Stock) stocksCache.elementAt(randomIndex);
	// // randomStock.modified = true;
	//
	// publishStocks();
	// }

	// private String getStocksLine() {
	// p("getStocksLine()");
	// BufferedReader br = null;
	// InputStream is = null;
	// String nextLine = "";
	//
	// // Read line from server
	// try {
	// is = new URL(pageURL).openStream();
	// br = new BufferedReader(new InputStreamReader(is));
	// boolean foundLine = false;
	// while (!foundLine) {
	// nextLine = br.readLine();
	// if (nextLine == null) {
	// return "";
	// }
	// foundLine = (nextLine.indexOf("details.asp?iid=14053&parent=aex") != -1);
	// }
	// } catch (Exception e) {
	// // e("could not open or read URL pageURL=" + pageURL + " ex=" + e);
	// return "";
	// } finally {
	// try {
	// if (is != null) is.close();
	// } catch (IOException ignore) {
	// }
	// }
	// return nextLine;
	// }
}

/*
 * $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
 * 
 * 
 */