pingpong.java

sea是一个基于seda模式的实现。这个设计模式将系统分为很多stage。每个stage分布不同的任务（基于线程池）。通过任务流的方式提高系统的效率。

/*
 * Copyright (c) 2003, The Regents of the University of California, through
 * Lawrence Berkeley National Laboratory (subject to receipt of any required
 * approvals from the U.S. Dept. of Energy). All rights reserved.
 */
package gov.lbl.dsd.sea.nio.demo;

import gov.lbl.dsd.sea.ExecutorFactory;
import gov.lbl.dsd.sea.Stage;
import gov.lbl.dsd.sea.StageManager;
import gov.lbl.dsd.sea.nio.AgentEventHandler;
import gov.lbl.dsd.sea.nio.NetAgent;
import gov.lbl.dsd.sea.nio.event.AdminRequest;
import gov.lbl.dsd.sea.nio.event.ChannelRequest;
import gov.lbl.dsd.sea.nio.event.ChannelResponse;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.charset.Charset;

/**
 * Simple non-blocking hello world client and server, echoing
 * messages back and forth.
 * <p>
 * Example server usage: fire-java gov.lbl.dsd.sea.nio.demo.PingPong server
 * 9000
 * <p>
 * Example client usage: fire-java gov.lbl.dsd.sea.nio.demo.PingPong client
 * localhost 9000
 * 
 * @author [e-mail]
 * @author $Author: gegles $
 * @version $Revision: 1.4 $, $Date: 2004/09/16 16:57:15 $
 */
public class PingPong extends AgentEventHandler {

	protected boolean isClient = true; // am I a client or a server?
	private static final Charset CHARSET = Charset.forName("US-ASCII");

	private static final org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory
			.getLog(PingPong.class);

	static public void main(String args[]) throws IOException {
		new PingPong(args);
	}

	public PingPong(String args[]) throws IOException {
		NetAgent agent = new NetAgent();

		String hostName = "localhost";
		int port = 9000;

		int k = -1;
		if (args.length > ++k) {
			isClient = args[k].equals("client");
		}
		if (isClient && args.length > ++k) {
			hostName = args[k];
		}
		if (args.length > ++k) {
			port = Integer.parseInt(args[k]);
		}

		ExecutorFactory execFactory = StageManager.QUEUED;
		if (args.length > ++k) {
			if (args[k].equals("single")) execFactory = StageManager.DIRECT;
		}		
		Stage myStage = new StageManager(execFactory).createStage(this).start();

		if (isClient) {
			agent.addConnectAddress(myStage, new InetSocketAddress(hostName,
					port));
		} else {
			agent.addListenPort(myStage, port);
		}

		agent.start();
	}

	protected void onClosed(ChannelResponse.Closed rsp) {
		log.info("*************got channel closed=" + rsp);
		if (rsp.getException() != null) {
			; // nothing special to be done
		}
		if (isClient) {
			this.shutDown(rsp);
		} else {
			// no problem; continueing
		}
		System.out.println("agent status=" + rsp.getAgent().toDebugString());
	}

	protected void onConnected(ChannelResponse.Connected rsp) {
		ByteBuffer buffer = CHARSET.encode("hello world");
		rsp.getAgent().enqueue(
				new ChannelRequest.Register(this.getStage(), rsp.getKey()
						.channel(), SelectionKey.OP_READ));
		rsp.getAgent().enqueue(
				new ChannelRequest.WriteData(rsp.getKey().channel(), buffer));
	}

	protected void onAccepted(ChannelResponse.Accepted rsp) {
		rsp.getAgent().enqueue(
				new ChannelRequest.Register(this.getStage(), rsp.getKey()
						.channel(), SelectionKey.OP_READ));
	}

	protected void onRead(ChannelResponse.Read rsp) {
		String str = CHARSET.decode(rsp.getBuffer()).toString();
		System.out.println("********************  read *********** " + str);
		
		try { // artificially slow down to better see progress on terminal
			Thread.sleep(500);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		
		ByteBuffer writeBuffer = CHARSET.encode(str + ".");
		
		// recycling to buffer pool is an optional optimization
		rsp.getAgent().getReadBufferPool().put(rsp.getBuffer());

		rsp.getAgent().enqueue(
				new ChannelRequest.WriteData(rsp.getKey().channel(), writeBuffer));
	}

	protected void onWrite(ChannelResponse.Write rsp) {
		rsp.getBuffer().flip();
		System.out.println("*************** fully wrote *********** "
				+ CHARSET.decode(rsp.getBuffer()).toString());

		// recycling to buffer pool is an optional optimization
		rsp.getAgent().getReadBufferPool().put(rsp.getBuffer());
	}

	protected void onRegistered(ChannelResponse.Registered rsp) {
		if (rsp.getInterestOps() == SelectionKey.OP_ACCEPT) {
			System.out.println("******** server started; now listening *****");
		}
	}

	private void shutDown(ChannelResponse rsp) {
		System.out.println("now shutting down...");
		this.getStage().stop();
		rsp.getAgent().enqueue(new AdminRequest.Stop());
	}

}