6.10 A Multiplexed Network Client

We end this chapter with Example 6-14, a complex class named HttpDownloadManager. As its name implies, this class is a client-side utility that manages any number of concurrent HTTP downloads on a background thread. To download something, just call the download( ) method, passing a java.net.URI and an optional Listener object to be notified when the download is complete or has aborted with an error. download( ) does not block: it returns a Download object immediately, and you can monitor the status of the download by polling the methods of this object. The data that is downloaded is not saved to a file, but is available from the getData( ) method of the Download object.

The Listener and Download interfaces are defined as inner classes of HttpDownloadManager. The Status class is another inner class: it is a type-safe enumeration of download states returned by Download.getStatus( ). Two other inner classes used in this example are DownloadImpl, the package-private concrete implementation of the Download interface, and Test, a simple test program that demonstrates the usage of the HttpDownloadManager.

HttpDownloadManager extends Thread, and the downloads are handled in this background thread. Multiple downloads can be handled concurrently (this is useful when the client has more bandwidth available than the servers which are being downloaded from) because the thread uses a Selector to multiplex the SocketChannel objects from which data is being read: the select( ) call wakes up whenever data is ready to be read from one of the pending downloads. Most of the interesting code is found in the run( ) method, which is the body of the background thread.

We've seen basic channel multiplexing code with Selector.select( ) in previous examples. This one demonstrates three new features. The first is the call to wakeup( ) in the download( ) method. A new channel cannot be registered with a Selector while that selector is blocked in a select( ) call in the background thread. So the download( ) method creates a DownloadImpl object containing all the information about the download, places this object in a synchronized list of pending downloads, and then calls wakeup( ) and returns the Download object to the caller. The wakeup( ) call in download( ) causes the background thread to stop blocking in select( ). In other examples, we've immediately checked the selectedKeys( ) of a Selector when its select( ) method returns. In this case, we first look at the pendingDownloads list and create and register a SocketChannel for any DownloadImpl objects found there.

The second new feature of interest in this example is that it performs asynchronous connection. Before an HTTP GET request can be sent to a web server, the client must connect to the server. Establishing a TCP connection over the Internet can sometimes take a second or two, and we don't want the background thread to block while this connection is set up. So when the thread detects a pending download, it creates a SocketChannel in an unconnected state. It then puts the channel into nonblocking mode and registers it with the Selector object, indicating that the Selector should monitor the channel for readiness to connect as well as readiness to read. Only after registering the channel does the thread call the connect( ) method and supply the address and port to connect to. Since the channel is nonblocking, connect( ) returns immediately. When the connection is ready, select( ) wakes up and the thread calls finishConnect( ) on the channel to complete the connection. (After completing the connection, the thread immediately sends the HTTP GET request across the channel. It assumes that the channel is writable and that the complete text of the request can be written quickly; if this assumption fails, the thread will end up busy-waiting while it repeatedly attempts to send the request to the server.)

The third new feature demonstrated by this example is that when it registers a SocketChannel with the Selector, it uses a three-argument version of the register( ) call to associate the DownloadImpl object with the SelectionKey for the channel. Then, when the key becomes connectable or readable, the background thread can retrieve the channel and the state object associated with the key.

Finally, this example also demonstrates the logging API of java.util.logging. We'll discuss logging in the subsection that follows the example code.

Example 6-14. HttpDownloadManager.java

package je3.nio; import java.io.*; import java.nio.*; import java.nio.channels.*; import java.nio.charset.*; import java.net.*; import java.util.*; import java.util.logging.*; /**  * This class manages asynchronous HTTP GET downloads and demonstrates  * non-blocking I/O with SocketChannel and Selector and also demonstrates  * logging with the java.util.logging package.  This example uses a number  * of inner classes and interfaces.  *   * Call download( ) for each HTTP GET request you want to issue.  You may  * optionally pass a Listener object that will be notified when the download  * terminates or encounters an exception.  download( ) returns a Download object  * which holds the downloaded bytes (including HTTP headers) and which allows  * you to poll the Status of the download.  Call release( ) when there are   * no more downloads.  */ public class HttpDownloadManager extends Thread {     // An enumerated type.  Values are returned by Download.getStatus( )     public static class Status {         // We haven't connected to the server yet         public static final Status UNCONNECTED = new Status("Unconnected");         // We're connected to the server, sending request or receiving response         public static final Status CONNECTED = new Status("Connected");         // Response has been received.  Response may have been an HTTP error         public static final Status DONE = new Status("Done");         // Something went wrong: bad hostname, for example.         public static final Status ERROR = new Status("Error");         private final String name;         private Status(String name) { this.name = name; }         public String toString( ) { return name; }     }     // Everything you need to know about a pending download     public interface Download {         public String getHost( );   // Hostname we're downloading from         public int getPort( );      // Defaults to port 80         public String getPath( );   // includes query string as well         public Status getStatus( ); // Status of the download         public byte[  ] getData( );   // Download data, including response headers         public int getHttpStatus( );// Only call when status is DONE     }     // Implement this interface if you want to know when a download completes     public interface Listener {         public void done(Download download);         public void error(Download download, Throwable throwable);     }     Selector selector;          // For multiplexing non-blocking I/O.     ByteBuffer buffer;          // A shared buffer for downloads     List pendingDownloads;      // Downloads that don't have a Channel yet     boolean released = false;   // Set when the release( ) method is called.     Logger log;                 // Logging output goes here     // The HTTP protocol uses this character encoding     static final Charset LATIN1 = Charset.forName("ISO-8859-1");     public HttpDownloadManager(Logger log) throws IOException {         if (log == null) log = Logger.getLogger(this.getClass( ).getName( ));         this.log = log;         selector = Selector.open( );                  // create Selector         buffer = ByteBuffer.allocateDirect(64*1024); // allocate buffer         pendingDownloads = Collections.synchronizedList(new ArrayList( ));         this.start( );                                // start thread     }     // Ask the HttpDownloadManager to begin a download.  Returns a Download     // object that can be used to poll the progress of the download.  The     // optional Listener object will be notified when the download completes     // or aborts.     public Download download(URI uri, Listener l)             throws IOException     {         if (released)            throw new IllegalStateException("Can't download( ) after release( )");         // Get info from the URI         String scheme = uri.getScheme( );         if (scheme == null || !scheme.equals("http"))             throw new IllegalArgumentException("Must use 'http:' protocol");         String hostname = uri.getHost( );         int port = uri.getPort( );         if (port == -1) port = 80; // Use default port if none specified         String path = uri.getRawPath( );         if (path == null || path.length( ) == 0) path = "/";         String query = uri.getRawQuery( );         if (query != null) path += "?" + query;         // Create a Download object with the pieces of the URL         Download download = new DownloadImpl(hostname, port, path, l);         // Add it to the list of pending downloads. This is a synchronized list         pendingDownloads.add(download);         // And ask the thread to stop blocking in the select( ) call so that         // it will notice and process this new pending Download object.         selector.wakeup( );         // Return the Download so that the caller can monitor it if desired.         return download;     }     public void release( ) {         released = true; // The thread will terminate when it notices the flag.         try { selector.close( ); } // This will wake the thread up         catch(IOException e) {             log.log(Level.SEVERE, "Error closing selector", e);         }     }     public void run( ) {         log.info("HttpDownloadManager thread starting.");         // The download thread runs until release( ) is called         while(!released) {             // The thread blocks here waiting for something to happen             try { selector.select( ); }             catch(IOException e) {                 // This should never happen.                 log.log(Level.SEVERE, "Error in select( )", e);                 return;             }             // If release( ) was called, the thread should exit.             if (released) break;             // If any new Download objects are pending, deal with them first             if (!pendingDownloads.isEmpty( )) {                 // Although pendingDownloads is a synchronized list, we still                 // need to use a synchronized block to iterate through its                 // elements to prevent a concurrent call to download( ).                 synchronized(pendingDownloads) {                     Iterator iter = pendingDownloads.iterator( );                     while(iter.hasNext( )) {                         // Get the pending download object from the list                         DownloadImpl download = (DownloadImpl)iter.next( );                         iter.remove( );  // And remove it.                         // Now begin an asynchronous connection to the                          // specified host and port.  We don't block while                         // waiting to connect.                         SelectionKey key = null;                         SocketChannel channel = null;                         try {                             // Open an unconnected channel                             channel = SocketChannel.open( );                             // Put it in non-blocking mode                             channel.configureBlocking(false);                             // Register it with the selector, specifying that                             // we want to know when it is ready to connect                             // and when it is ready to read.                             key = channel.register(selector,                                                    SelectionKey.OP_READ |                                                     SelectionKey.OP_CONNECT,                                                    download);                             // Create the web server address                             SocketAddress address =                                  new InetSocketAddress(download.host,                                                       download.port);                             // Ask the channel to start connecting                             // Note that we don't send the HTTP request yet.                             // We'll do that when the connection completes.                             channel.connect(address);                         }                         catch(Exception e) {                             handleError(download, channel, key, e);                         }                     }                 }             }             // Now get the set of keys that are ready for connecting or reading             Set keys = selector.selectedKeys( );             if (keys == null) continue; // bug workaround; should not be needed             // Loop through the keys in the set             for(Iterator i = keys.iterator( ); i.hasNext( ); ) {                 SelectionKey key = (SelectionKey)i.next( );                 i.remove( );  // Remove the key from the set before handling                 // Get the Download object we attached to the key                 DownloadImpl download = (DownloadImpl) key.attachment( );                 // Get the channel associated with the key.                 SocketChannel channel = (SocketChannel)key.channel( );                 try {                     if (key.isConnectable( )) {                           // If the channel is ready to connect, complete the                         // connection and then send the HTTP GET request to it.                         if (channel.finishConnect( )) {                             download.status = Status.CONNECTED;                             // This is the HTTP request we send                             String request =                                 "GET " + download.path + " HTTP/1.1\r\n" +                                 "Host: " + download.host + "\r\n" +                                 "Connection: close\r\n" +                                 "\r\n";                             // Wrap in a CharBuffer and encode to a ByteBuffer                             ByteBuffer requestBytes =                                 LATIN1.encode(CharBuffer.wrap(request));                             // Send the request to the server.  If the bytes                             // aren't all written in one call, we busy loop!                             while(requestBytes.hasRemaining( ))                                 channel.write(requestBytes);                             log.info("Sent HTTP request: " + download.host +                                       ":" + download.port + ": " + request);                         }                     }                     if (key.isReadable( )) {                         // If the key indicates that there is data to be read,                         // then read it and store it in the Download object.                         int numbytes = channel.read(buffer);                                                  // If we read some bytes, store them, otherwise                         // the download is complete and we need to note this                         if (numbytes != -1) {                             buffer.flip( );  // Prepare to drain the buffer                             download.addData(buffer); // Store the data                             buffer.clear( ); // Prepare for another read                             log.info("Read " + numbytes + " bytes from " +                                      download.host + ":" + download.port);                         }                         else {                             // If there are no more bytes to read                             key.cancel( );     // We're done with the key                             channel.close( );  // And with the channel.                             download.status = Status.DONE;                              if (download.listener != null)  // notify listener                                 download.listener.done(download);                             log.info("Download complete from " +                                      download.host + ":" + download.port);                         }                     }                 }                 catch (Exception e) {                     handleError(download, channel, key, e);                 }             }         }         log.info("HttpDownloadManager thread exiting.");     }     // Error-handling code used by the run( ) method:      // set status, close channel, cancel key, log error, notify listener.     void handleError(DownloadImpl download, SocketChannel channel,                      SelectionKey key, Throwable throwable)     {         download.status = Status.ERROR;         try {if (channel != null) channel.close( );} catch(IOException e) {  }         if (key != null) key.cancel( );         log.log(Level.WARNING,                 "Error connecting to or downloading from " + download.host +                 ":" + download.port,                 throwable);         if (download.listener != null)             download.listener.error(download, throwable);     }     // This is the Download implementation we use internally.     static class DownloadImpl implements Download {         final String host;     // Final fields are immutable for thread-safety         final int port;         final String path;         final Listener listener;         volatile Status status; // Volatile fields may be changed concurrently         volatile byte[  ] data = new byte[0];         DownloadImpl(String host, int port, String path, Listener listener) {             this.host = host;             this.port = port;             this.path = path;             this.listener = listener;             this.status = Status.UNCONNECTED;  // Set initial status         }         // These are the basic getter methods         public String getHost( ) { return host; }         public int getPort( ) { return port; }         public String getPath( ) { return path; }         public Status getStatus( ) { return status; }         public byte[  ] getData( ) { return data; }         /**          * Return the HTTP status code for the download.          * Throws IllegalStateException if status is not Status.DONE          */         public int getHttpStatus( ) {             if (status != Status.DONE) throw new IllegalStateException( );             // In HTTP 1.1, the return code is in ASCII bytes 10-12.             return                 (data[9] - '0') * 100 +                 (data[10]- '0') * 10 +                 (data[11]- '0') * 1;         }         // Used internally when we read more data.         // This should use a larger buffer to prevent frequent re-allocation.         void addData(ByteBuffer buffer) {             assert status == Status.CONNECTED;  // only called during download             int oldlen = data.length;           // How many existing bytes             int numbytes = buffer.remaining( );  // How many new bytes             int newlen = oldlen + numbytes;             byte[  ] newdata = new byte[newlen];  // Create new array             System.arraycopy(data, 0, newdata, 0, oldlen); // Copy old bytes             buffer.get(newdata, oldlen, numbytes);         // Copy new bytes             data = newdata;                     // Save new array         }     }     // This class demonstrates a simple use of HttpDownloadManager.     public static class Test {         static int completedDownloads = 0;         public static void main(String args[  ])             throws IOException, URISyntaxException         {             // With a -v argument, our logger will display lots of messages             final boolean verbose = args[0].equals("-v");             int firstarg = 0;             Logger logger = Logger.getLogger(Test.class.getName( ));             if (verbose) {                 firstarg = 1;                 logger.setLevel(Level.INFO);             }             else                       // regular output                 logger.setLevel(Level.WARNING);                          // How many URLs are on the command line?             final int numDownloads = args.length - firstarg;             // Create the download manager             final HttpDownloadManager dm = new HttpDownloadManager(logger);             // Now loop through URLs and call download( ) for each one             // passing a listener object to receive notifications             for(int i = firstarg; i < args.length; i++) {                 URI uri = new URI(args[i]);                 dm.download(uri,                      new Listener( ) {                             public void done(Download d) {                                 System.err.println("DONE: " + d.getHost( ) +                                                    ": " + d.getHttpStatus( ));                                 // If all downloads are complete, we're done                                 // with the HttpDownloadManager thread.                                 if (++completedDownloads == numDownloads)                                     dm.release( );                             }                             public void error(Download d, Throwable t) {                                 System.err.println(d.getHost( ) + ": " + t);                                 if (++completedDownloads == numDownloads)                                     dm.release( );                             }                         });             }         }     } }

6.10.1 Logging in HttpDownloadManager

HttpDownloadManager uses a Logger object to log informational and error messages. It demonstrates the info( ) and log( ) methods of Logger; see the Logger documentation for descriptions of many other logging methods. HttpDownloadManager can use a Logger passed to it, or it can obtain its own. There are examples of obtaining a Logger object in both the constructor method and the inner Test class. The Test class also implements a -v "verbose" switch to demonstrate how to set the logging threshold of a Logger. Informational logging messages will be discarded unless the -v option is specified.

The java.util.logging package allows flexible runtime configuration of how logging is done. In most installations, the default is to print logging messages to the console. See the file jre/lib/logging.properties for the default configuration on your installation. You can override this default configuration with your own properties file: define the property java.util.logging.config.file with the -D option when you start the Java VM. For example, to run the HttpDownloadManager test program using a logging configuration specified in a file named log.props, you'd use a command line like this one (it has been word wrapped to fit on the page; you'd type it on one line):

java -Djava.util.logging.config.file=log.props je3.nio.HttpDownloadManager\$Test -v ...urls here...

Note that we use the -v switch so that the program actually generates log messages. Without this switch, you'd have to purposely specify bad URLs so that the program would log some errors. The following listing is a sample logging configuration file, which you'll find, along with the HttpDownloadManager source code, in a file named log.props:

# # A logging properties file for the HttpDownloadManager example # See also jre/lib/logging.properties in your Java installation. # Use this file by specifying it as the value of a property named # java.util.logging.config.file.  For example: # # java -Djava.util.logging.config.file je3.nio.Http... # # This property specifies the default logging level for log messages # sent by our program.  Note, however, that if you run the  # HttpDownloadManager\$Test class, this property will be overridden by # the presence or absence of the -v option. je3.nio.HttpDownloadManager.level: INFO # This property says where output should go.  The default configuration # is for it to go to the console.  This property sends log messages to # a FileHandler configured below instead handlers=java.util.logging.FileHandler # These properties configure the FileHandler # See java.util.logging.FileHandler for other available properties java.util.logging.FileHandler.pattern = %h/java%u.log java.util.logging.FileHandler.formatter = java.util.logging.XMLFormatter