generichandler.java

java网络编程方面的源码,其中有一个整合的聊天室,比较不错,建议大家下载练习,配合java网络编程技术内幕看

/*
 * Java Network Programming, Second Edition
 * Merlin Hughes, Michael Shoffner, Derek Hamner
 * Manning Publications Company; ISBN 188477749X
 *
 * http://nitric.com/jnp/
 *
 * Copyright (c) 1997-1999 Merlin Hughes, Michael Shoffner, Derek Hamner;
 * all rights reserved; see license.txt for details.
 */

import java.io.*;
import java.util.*;

public class GenericHandler extends Thread {
  static private int handlerNumber;
  static private synchronized int nextHandlerNum ()
    { return handlerNumber ++; }

  protected Hashtable routes;
  protected InputStream in;
  protected OutputStream out;

  public GenericHandler (Hashtable routes, InputStream in, OutputStream out) {
    super ("GenericHandler-" + nextHandlerNum ());
    this.routes = routes;
    this.in = in;
    this.out = out;
  }

  protected String name;
  
  public void run () {
    try {
      DataInputStream dataIn = new DataInputStream (in);
      name = dataIn.readUTF ();
      accept (name);
    } catch (IOException ex) {
      ex.printStackTrace ();
    } finally {
      try {
        out.close ();
      } catch (IOException ex) {
        ex.printStackTrace ();
      }
    }
  }

  protected MessageOutput myself;
  
  protected void accept (String name) throws IOException {
    Queue queue = new Queue ();
    boolean registered = false;
    synchronized (routes) {
      if (!routes.containsKey (name)) {
        registered = true;
        routes.put (name, myself = new QueueOutputStream (queue));
      }
    }
    try {
      new DataOutputStream (out).writeBoolean (registered);
      out.flush ();
      if (registered) {
        execute (queue);
      }
    } finally {
      if (registered)
        routes.remove (name);
    }
  }

  protected void execute (Queue queue) throws IOException {
    MessageInput queueIn = new QueueInputStream (queue);
    MessageOutput messageOut = new MessageOutputStream (out);
    GenericMessageCopier copier =
      new GenericMessageCopier (this, queueIn, messageOut);
    try {
      copier.start ();
      route ();
    } finally {
      copier.finish ();
    }
  }

  protected void route () throws IOException {
    MessageInputStream messageIn = new MessageInputStream (in);
    RoutingInputStream routingIn = new RoutingInputStream (messageIn);
    while (!Thread.interrupted ()) {
      routingIn.receive ();
      byte[] buffer = new byte[routingIn.available ()];
      routingIn.readFully (buffer);
      String[] targets = routingIn.getTargets ();
      if (targets == null)
        broadcast (buffer);
      else
        multicast (buffer, targets);
    }
  }

  protected void broadcast (byte[] buffer) throws IOException {
    Enumeration dsts = ((Hashtable) routes.clone ()).elements ();
    while (dsts.hasMoreElements ()) {
      MessageOutput messageOut = (MessageOutput) dsts.nextElement ();
      send (buffer, messageOut);
    }
  }

  protected void multicast (byte[] buffer, String[] targets) throws IOException {
    for (int j = 0; j < targets.length; ++ j) {
      MessageOutput messageOut = (MessageOutput) routes.get (targets[j]);
      if (messageOut != null)
        send (buffer, messageOut);
    }
  }

  protected void send (byte[] buffer, MessageOutput messageOut)
    throws IOException {
    if (messageOut != myself) {
      synchronized (messageOut) {
        messageOut.write (buffer);
        messageOut.send ();
      }
    }
  }
}