pushletclient.java

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

			// Stop and report error.
			warn("doControl() exception", t);
			throw new PushletException(" error parsing response from" + controlURL, t);
		}
	}

	/**
	 * Util: print.
	 */
	private void p(String s) {
		if (debug) {
			System.out.println("[PushletClient] " + s);
		}
	}

	/**
	 * Util: warn.
	 */
	private void warn(String s) {
		warn(s, null);
	}

	/**
	 * Util: warn with exception.
	 */
	private void warn(String s, Throwable t) {
		System.err.println("[PushletClient] - WARN - " + s + " ex=" + t);

		if (t != null) {
			t.printStackTrace();
		}
	}

	/**
	 * Internal listener for the Pushlet data channel.
	 */
	private class DataEventListener implements Runnable {
		/**
		 * Client's listener that gets called back on events.
		 */
		private PushletClientListener listener;

		/**
		 * Receiver receiveThread.
		 */
		private Thread receiveThread = null;
		private Reader reader;
		private String refreshURL;
		private String listenURL;

		public DataEventListener(PushletClientListener aListener, String aListenURL) {
			listener = aListener;
			listenURL = aListenURL;
		}

		public void start() {
			// All ok: start a receiver receiveThread
			receiveThread = new Thread(this);
			receiveThread.start();

		}

		/**
		 * Stop listening; may restart later with start().
		 */
		public void stop() {
			p("In stop()");
			bailout();
		}

		/**
		 * Receive event objects from server and callback listener.
		 */
		public void run() {
			p("Start run()");
			try {
				while (receiveThread != null && receiveThread.isAlive()) {
					// Connect to server
					reader = openURL(listenURL);

					synchronized (this) {
						// Inform the calling thread we're ready to receive events.
						// Suggestion by Jeff Nowakowski 29.oct.2006
						this.notify();
					}

					// Get events while we're alive.
					while (receiveThread != null && receiveThread.isAlive()) {
						Event event = null;
						try {
							// p("Getting event...");
							// Get next event from server
							event = EventParser.parse(reader);
							
							//就是这。每次OnData 和 OnHeartBeat时都输出event
							//p("Event received " + event);
						} catch (Throwable t) {

							// Stop and report error.
							// warn("Stop run() on exception", t);
							if (listener != null) {
								listener.onError("exception during receive: " + t);
							}

							break;
						}

						// Handle event by calling listener
						if (event != null && listener != null) {
							// p("received: " + event.toXML());
							String eventType = event.getEventType();
							if (eventType.equals(E_HEARTBEAT)) {
								listener.onHeartbeat(event);
							} else if (eventType.equals(E_DATA)) {
								listener.onData(event);
							} else if (eventType.equals(E_JOIN_LISTEN_ACK)) {
								id = event.getField(P_ID);
							} else if (eventType.equals(E_LISTEN_ACK)) {
								p("Listen ack ok");
							} else if (eventType.equals(E_REFRESH_ACK)) {
								// ignore
							} else if (eventType.equals(E_ABORT)) {
								listener.onAbort(event);
								listener = null;
								break;
							} else if (eventType.equals(E_REFRESH)) {
								refresh(event);
							} else {
								warn("unsupported event type received: " + eventType);
							}
						}
					}
				}
			} catch (Throwable t) {
				warn("Exception in run() ", t);
				// bailout();
			}
		}

		private void disconnect() {
			p("start disconnect()");
			if (reader != null) {
				try {
					// this blocks, find another way
					// reader.close();
					p("Closed reader ok");
				} catch (Exception ignore) {
				} finally {
					reader = null;
				}
			}
			p("end disconnect()");
		}

		/**
		 * Stop receiver receiveThread.
		 */
		public void stopThread() {
			p("In stopThread()");

			// Keep a reference such that we can kill it from here.
			Thread targetThread = receiveThread;

			receiveThread = null;

			// This should stop the main loop for this receiveThread.
			// Killing a receiveThread on a blcing read is tricky.
			// See also http://gee.cs.oswego.edu/dl/cpj/cancel.html
			if ((targetThread != null) && targetThread.isAlive()) {

				targetThread.interrupt();

				try {

					// Wait for it to die
					targetThread.join(500);
				} catch (InterruptedException ignore) {
				}

				// If current receiveThread refuses to die,
				// take more rigorous methods.
				if (targetThread.isAlive()) {

					// Not preferred but may be needed
					// to stop during a blocking read.
					targetThread.stop();

					// Wait for it to die
					try {
						targetThread.join(500);
					} catch (Throwable ignore) {
					}
				}

				p("Stopped receiveThread alive=" + targetThread.isAlive());

			}
		}

		/**
		 * Stop listening on stream from server.
		 */
		public void bailout() {
			p("In bailout()");
			stopThread();
			disconnect();
		}

		/**
		 * Handle refresh, by pausing.
		 */
		private void refresh(Event aRefreshEvent) throws PushletException {
			try {
				// Wait for specified time.
				Thread.sleep(Long.parseLong(aRefreshEvent.getField(P_WAIT)));
			} catch (Throwable t) {
				warn("abort while refresing");
				refreshURL = null;
				return;
			}

			// If stopped during sleep, don't proceed
			if (receiveThread == null) {
				return;
			}

			// Create url to refresh
			refreshURL = pushletURL
					+ "?" + P_ID + "=" + id
					+ "&" + P_EVENT + "=" + E_REFRESH
					;

			if (reader != null) {
				try {
					reader.close();

				} catch (IOException ignore) {

				}
				reader = null;
			}

			reader = openURL(refreshURL);
		}
	}

	/**
	 * Authenticator
	 */
	private static class HTTPAuthenticateProxy extends Authenticator {

		/**
		 * Contributed by Dele Olajide
		 * See http://groups.yahoo.com/group/pushlet/message/634
		 */

		private String thePassword = "";
		private String theUser = "";

		public HTTPAuthenticateProxy(String username, String password) {

			thePassword = password;
			theUser = username;
		}

		protected PasswordAuthentication getPasswordAuthentication() {
			// System.out.println("[HttpAuthenticateProxy] Username = " + theUser);
			// System.out.println("[HttpAuthenticateProxy] Password = " + thePassword);

			return new PasswordAuthentication(theUser, thePassword.toCharArray());
		}

	}

}

/*
 * $Log: PushletClient.java,v $
 * Revision 1.18  2007/11/10 13:52:47  justb
 * make startDataEventListener method protected to allow overriding
 *
 * Revision 1.17  2006/10/29 16:47:57  justb
 * included patch from Jeff Nowakowski: wait until listener thread runs
 *
 * Revision 1.16  2005/05/06 20:08:20  justb
 * client enhancements
 *
 * Revision 1.15  2005/03/27 17:42:27  justb
 * enhancements
 *
 * Revision 1.14  2005/03/25 23:54:04  justb
 * *** empty log message ***
 *
 * Revision 1.13  2005/02/28 16:59:40  justb
 * fixes for leave and disconnect
 *
 * Revision 1.12  2005/02/28 15:57:54  justb
 * added SimpleListener example
 *
 * Revision 1.11  2005/02/21 12:31:44  justb
 * added proxy contribution from Dele Olajide
 *
 * Revision 1.10  2005/02/20 13:05:32  justb
 * removed the Postlet (integrated in Pushlet protocol)
 *
 * Revision 1.9  2005/02/18 10:07:23  justb
 * many renamings of classes (make names compact)
 *
 * Revision 1.8  2005/02/18 09:54:12  justb
 * refactor: rename Publisher Dispatcher and single Subscriber class
 *
 * Revision 1.7  2005/02/15 15:46:30  justb
 * client API improves
 *
 * Revision 1.6  2005/02/15 13:28:56  justb
 * first quick rewrite adapt for v2 protocol
 *
 * Revision 1.5  2004/10/25 21:23:44  justb
 * *** empty log message ***
 *
 * Revision 1.4  2004/10/24 13:52:51  justb
 * small fixes in client lib
 *
 * Revision 1.3  2004/10/24 12:58:18  justb
 * revised client and test classes for new protocol
 *
 * Revision 1.2  2004/09/03 22:35:37  justb
 * Almost complete rewrite, just checking in now
 *
 * Revision 1.1  2004/03/10 20:14:17  justb
 * renamed all *JavaPushletClient* to *PushletClient*
 *
 * Revision 1.10  2004/03/10 15:45:55  justb
 * many cosmetic changes
 *
 * Revision 1.9  2003/08/17 20:30:20  justb
 * cosmetic changes
 *
 * Revision 1.8  2003/08/15 08:37:40  justb
 * fix/add Copyright+LGPL file headers and footers
 *
 *
 */