peer.java

利用hashtable实现网络广播

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

public class Peer {
  public final static long BCAST_PEER = -1;
  public final static long BCAST_INIT = 0;
  public final static long BCAST_MIN = 1;

  static Vector Peers = new Vector();
  static Hashtable PeerTab = new Hashtable();
  static String my_addr;

  public static void main( String [] args ) {
    int port = 0;
    int join_port = 0;
    int len;
    Socket s;
    ServerSocket ss;
    InetAddress addr;
    String st, src, cmd, msg, join;
    Scanner sc;

    try {
      if( args.length < 3 ) {
        sc = new Scanner( System.in );
        System.out.print( "Enter port to listen on: " );
        if( sc.hasNextInt() ) {
	  port = sc.nextInt();
          sc.nextLine();
        }

        System.out.print( "Enter peer's address to join to: " );
        join = sc.nextLine();
        if( join.length() > 1 ) {
          System.out.print( "Enter peer's port to join to: " );
          if( sc.hasNextInt() ) {
            join_port = sc.nextInt();
            sc.nextLine();
          }
        }
      } else {
        port = Integer.parseInt( args[0] );
        join = args[1];
        if( join.length() > 1 ) {
          join_port = Integer.parseInt( args[2] );
        }
      }

      if( ( port < 1024 ) || ( port > 32767 ) ) {
        System.out.println( "Bad port specified" );
        return;
      }

      ss = new ServerSocket( port );
      addr = InetAddress.getLocalHost();
      my_addr = addr.getHostAddress() + ":" + port;
      System.out.println( "I am at: " + my_addr );

      if( join.length() > 1 ) {
         if( ( join_port < 1024 ) || ( join_port > 32767 ) ) {
           System.out.println( "Bad join port specified" );
           return;
         }

         // Join the network
         System.out.println( "Joining" );
         join( join, join_port );
      }

      // Enter listen loop
      System.out.println( "Listening" );
      while( true ) {
        try {
          s = ss.accept();
          sc = new Scanner( read( s ) );
          cmd = sc.next();
          src = sc.next();

          if( cmd.equals( "JOIN" ) ) {
            welcome( s, src );
          } else if( cmd.equals( "BCAST" ) ) {
            bcast( sc.nextLong(), sc.nextLine(), src );
          }
          s.close();
        } catch ( Exception e ) { 
          System.out.println( "Continuing after exception: " + e );
        }
      }
    } catch ( Exception e ) {
      System.out.println( e );
    }
  }

  static void join( String host, int port ) {
    String st;
    Socket s;
    Scanner sc;

    try {
      s = new Socket( host, port );
      write( s, "JOIN " + my_addr );
      st = read( s );
      if( st != null ) {
        sc = new Scanner( st );
        while( sc.hasNext() ) {
          addPeer( sc.next() );
        }
      }
      s.close();
    } catch ( Exception e ) {
      System.out.println( e );
      System.exit( 0 );
    }
  }

  static void welcome( Socket s, String src ) {
    int i;
    int size = Peers.size();
    String peer;
    String resp = my_addr;

    for( i = 0; i < size; i++ ) {
      peer = (String) Peers.elementAt( i );
      if( !peer.equals( src ) ) {
        resp = resp + " " + peer;
      }
    }
    write( s, resp );
    bcast( BCAST_PEER, src, src );
  }

  static boolean addPeer( String peer ) {
    if( PeerTab.get( peer ) == null ) {
      Peers.add( peer );
      PeerTab.put( peer, new SockAddr( peer ) );
      return true;
    }
    return false;
  }

  public static boolean write( Socket s, String st ) {
    OutputStream out;

    try {
      out = s.getOutputStream();
      out.write( st.getBytes( "US-ASCII" ) );
      out.write( 0 );
      out.flush();
      return true;
    } catch ( Exception e ) {
      System.out.println( e );
      return false;
    }
  }

  public static String read( Socket s ) {
    InputStream in;
    byte [] b = new byte[65536];
    int len;

    try {
      in = s.getInputStream();
      for( len = 0; in.read( b, len, 1 ) == 1; len++ ) {
        if( b[len] == 0 ) {
          return new String( b, 0, len, "US-ASCII" );
        }
      }
    } catch ( Exception e ) {
      System.out.println( e );
    }
    return null;
  }

  static void bcast( long time, String msg, String src ) {
    Thread t;
    String peer, payload;
    int i, size;
    SockAddr sa;

    if( msg.charAt( 0 ) == ' ' ) {
      msg = msg.substring( 1 );
    }

    if( time == BCAST_PEER ) { 
      if( !addPeer( msg ) ) {
        return;
      }
    } else {
      time = MsgTab.add( time, msg );
      if( time < BCAST_MIN ) {
        return;
      }
      System.out.println( "[" + my_addr + "] " + msg );
    } 

    payload = "BCAST " + my_addr + " " + time + " " + msg;
    size = Peers.size();
    for( i = 0; i < size; i++ ) {
      peer = (String) Peers.elementAt( i );
      if( !peer.equals( src ) ) {
        sa = (SockAddr) PeerTab.get( peer );
        if( ( sa != null ) && sa.active ) {
          t = new Thread( new Sender( sa, payload ) );
          t.start();
        }
      }
    }
  }

  private static class MsgTab {
    final static long EXPIRY = 60000; // 1 minute expiry
    static MsgTab head;
    static MsgTab tail;
    static Hashtable map = new Hashtable();

    MsgTab next;
    String payload;
    long time;

    public static long add( long time, String msg ) {
      Date d;
      MsgTab m;
      long expiry;
      String payload = time + " " + msg;

      if( ( time != BCAST_INIT ) && map.get( payload ) != null ) {
        return 0;
      }

      d = new Date();
      m = new MsgTab();
      m.time = d.getTime();
      if( time == BCAST_INIT ) {
        time = m.time;
        m.payload = time + " " + msg;
      } else {
        m.payload = payload;
      }
  
      if( head == null ) {
        head = m;
      } else {
        tail.next = m;
      }
      tail = m;
      map.put( m.payload, m );
  
      expiry = m.time - EXPIRY;
      while( ( head != null ) && ( head.time < expiry ) ) {
        map.remove( head.payload );
        head = head.next;
      }
    
      return time;
    }
  }

  static class SockAddr {
    String addr = null;
    int port = 0;
    boolean active = false;

    SockAddr( String peer ) {
      String [] parts;

      try {
        parts = peer.split( ":" );
        addr = parts[0];
        port = Integer.parseInt( parts[1] );
        active = true;
      } catch ( Exception e ) {
        System.out.println( e );
        active = false;
      }
    } 
  }

  static class Sender implements Runnable {
    SockAddr sockaddr;
    String payload;
   
    Sender( SockAddr sa, String p ) {
      sockaddr = sa;
      payload = p;
    }

    public void run() {
      Socket s;

      try {
        s = new Socket( sockaddr.addr, sockaddr.port );
        write( s, payload );
        s.close();
      } catch ( Exception e ) {
        sockaddr.active = false;
      }
    }
  }
}