src/java/org/opensrf/net/http/HttpConnection.java

   1 package org.opensrf.net.http;
   2
   3 import java.net.URL;
   4 import java.net.MalformedURLException;
   5 import java.util.Queue;
   6 import java.util.concurrent.ConcurrentLinkedQueue;
   7 import org.opensrf.*;
   8 import org.opensrf.util.*;
   9
  10
  11 /**
  12  * Manages connection parameters and thread limiting for opensrf json gateway connections.
  13  */
  14
  15 public class HttpConnection {
  16
  17     /** Compiled URL object */
  18     protected URL url;
  19     /** Number of threads currently communicating with the server */
  20     protected int activeThreads;
  21     /** Queue of pending async requests */
  22     protected Queue<HttpRequest> pendingThreadQueue;
  23     /** maximum number of actively communicating threads allowed */
  24     protected int maxThreads = 10;
  25
  26     public HttpConnection(String fullUrl) throws java.net.MalformedURLException {
  27         activeThreads = 0;
  28         pendingThreadQueue = new ConcurrentLinkedQueue();
  29         url = new URL(fullUrl);
  30     }
  31
  32     /**
  33      * Maximun number of threads allowed to communicate with the server.
  34      */
  35     public int getMaxThreads() {
  36         return maxThreads;
  37     }
  38
  39     /**
  40      * Set the maximum number of actively communicating (async) threads allowed.
  41      *
  42      * This has no effect on synchronous communication.
  43      */
  44     public void setMaxThreads(int max) {
  45         maxThreads = max;
  46     }
  47
  48     /**
  49      * Launches or queues an asynchronous request.
  50      *
  51      * If the maximum active thread count has not been reached,
  52      * start a new thread and use it to send and receive the request.
  53      * The response is passed to the request's HttpRequestHandler
  54      * onComplete().  After complete, if the number of active threads
  55      * is still lower than the max, one request will be pulled (if
  56      * present) from the async queue and fired.
  57      *
  58      * If there are too many active threads, the main request is
  59      * pushed onto the async queue for later processing
  60      */
  61     protected void manageAsyncRequest(final HttpRequest request) {
  62
  63         if (activeThreads >= maxThreads) {
  64             pendingThreadQueue.offer(request);
  65             return;
  66         }
  67
  68         activeThreads++;
  69
  70          //Send the request receive the response, fire off the next
  71          //thread if necessary, then pass the result to the handler
  72         Runnable r = new Runnable() {
  73             public void run() {
  74                 Object response;
  75
  76                 try {
  77                     request.send();
  78                     while ((response = request.recv()) != null)
  79                         request.handler.onResponse(request, response);
  80
  81                     request.handler.onComplete(request);
  82
  83                 } catch (Exception ex) {
  84                     request.handler.onError(request, ex);
  85
  86                 } finally {
  87                     // server communication has completed
  88                     activeThreads--;
  89                 }
  90
  91                 if (activeThreads < maxThreads) {
  92                     try {
  93                         manageAsyncRequest(pendingThreadQueue.remove());
  94                     } catch (java.util.NoSuchElementException ex) {
  95                         // may have been gobbled by another thread
  96                     }
  97                 }
  98             }
  99         };
 100
 101         new Thread(r).start();
 102     }
 103 }
 104
 105