android-ibc-forum

RSSLoader.java (13324B)

---

 /*
  * Copyright (C) 2011 A. Horn
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.mcsoxford.rss;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Asynchronous loader for RSS feeds. RSS feeds can be loaded in FIFO order or
  * based on priority. Objects of this type can be constructed with one of the
  * provided static methods:
  * <ul>
  * <li>{@link #fifo()}</li>
  * <li>{@link #fifo(int)}</li>
  * <li>{@link #priority()}</li>
  * <li>{@link #priority(int)}</li>
  * </ul>
  * 
  * Completed RSS feed loads can be retrieved with {@link RSSLoader#take()},
  * {@link RSSLoader#poll()} or {@link RSSLoader#poll(long, TimeUnit)}.
  * 
  * <p>
  * <b>Usage Example</b>
  * 
  * Suppose you want to load an array of RSS feed URIs concurrently before
  * retrieving the results one at a time. You could write this as:
  * 
  * <pre>
  * {@code 
  *  void fetchRSS(String[] uris) throws InterruptedException {
  *     RSSLoader loader = RSSLoader.fifo();
  *     for (String uri : uris) {
  *       loader.load(uri);
  *     }
  *     
  *     Future&lt;RSSFeed&gt; future;
  *     RSSFeed feed;
  *     for (int i = 0; i &lt; uris.length; i++) {
  *       future = loader.take();
  *       try {
  *         feed = future.get();
  *         use(feed);
  *       } catch (ExecutionException ignore) {}
  *     }
  * }}
  * </pre>
  *
  * </p>
  * 
  * @author A. Horn
  */
 public class RSSLoader {
 
   /**
    * Human-readable name of the thread loading RSS feeds
    */
   private final static String DEFAULT_THREAD_NAME = "Asynchronous RSS feed loader";
 
   /**
    * Arrange incoming load requests on this queue.
    */
   private final BlockingQueue<RSSFuture> in;
 
   /**
    * Once the an RSS feed has completed loading, place the result on this queue.
    */
   private final BlockingQueue<RSSFuture> out;
 
   /**
    * Flag changes are visible after operations on {@link #in} queue.
    */
   private boolean stopped;
 
   /**
    * Create an object which can load RSS feeds asynchronously in FIFO order.
    * 
    * @see #fifo(int)
    */
   public static RSSLoader fifo() {
     return new RSSLoader(new LinkedBlockingQueue<RSSFuture>());
   }
 
   /**
    * Create an object which can load RSS feeds asynchronously in FIFO order.
    * 
    * @param capacity
    *          expected number of URIs to be loaded at a given time
    */
   public static RSSLoader fifo(int capacity) {
     return new RSSLoader(new LinkedBlockingQueue<RSSFuture>(capacity));
   }
 
   /**
    * Create an object which can load RSS feeds asynchronously based on priority.
    * 
    * @see #priority(int)
    */
   public static RSSLoader priority() {
     return new RSSLoader(new PriorityBlockingQueue<RSSFuture>());
   }
 
   /**
    * Create an object which can load RSS feeds asynchronously based on priority.
    * 
    * @param capacity
    *          expected number of URIs to be loaded at a given time
    */
   public static RSSLoader priority(int capacity) {
     return new RSSLoader(new PriorityBlockingQueue<RSSFuture>(capacity));
   }
 
   /**
    * Instantiate an object which can load RSS feeds asynchronously. The provided
    * {@link BlockingQueue} implementation determines the load behaviour.
    * 
    * @see LinkedBlockingQueue
    * @see PriorityBlockingQueue
    */
   RSSLoader(BlockingQueue<RSSFuture> in) {
     this.in = in;
     this.out = new LinkedBlockingQueue<RSSFuture>();
 
     // start separate thread for loading of RSS feeds
     new Thread(new Loader(new RSSReader()), DEFAULT_THREAD_NAME).start();
   }
 
   /**
    * Returns {@code true} if RSS feeds are currently being loaded, {@code false}
    * otherwise.
    */
   public boolean isLoading() {
     // order of conjuncts matters because of happens-before relationship
     return !in.isEmpty() && !stopped;
   }
 
   /**
    * Stop thread after finishing loading pending RSS feed URIs. If this loader
    * has been constructed with {@link #priority()} or {@link #priority(int)},
    * only RSS feed loads with priority strictly greater than seven (7) are going
    * to be completed.
    * <p>
    * Subsequent invocations of {@link #load(String)} and
    * {@link #load(String, int)} return {@code null}.
    */
   public void stop() {
     // flag writings happen-before enqueue
     stopped = true;
     in.offer(SENTINEL);
   }
 
   /**
    * Loads the specified RSS feed URI asynchronously. If this loader has been
    * constructed with {@link #priority()} or {@link #priority(int)}, then a
    * default priority of three (3) is used. Otherwise, RSS feeds are loaded in
    * FIFO order.
    * <p>
    * Returns {@code null} if the RSS feed URI cannot be scheduled for loading
    * due to resource constraints or if {@link #stop()} has been previously
    * called.
    * <p>
    * Completed RSS feed loads can be retrieved by calling {@link #take()}.
    * Alternatively, non-blocking polling is possible with {@link #poll()}.
    * 
    * @param uri
    *          RSS feed URI to be loaded
    * 
    * @return Future representing the RSS feed scheduled for loading,
    *         {@code null} if scheduling failed
    */
   public Future<RSSFeed> load(String uri) {
     return load(uri, RSSFuture.DEFAULT_PRIORITY);
   }
 
   /**
    * Loads the specified RSS feed URI asynchronously. For the specified priority
    * to determine the relative loading order of RSS feeds, this loader must have
    * been constructed with {@link #priority()} or {@link #priority(int)}.
    * Otherwise, RSS feeds are loaded in FIFO order.
    * <p>
    * Returns {@code null} if the RSS feed URI cannot be scheduled for loading
    * due to resource constraints or if {@link #stop()} has been previously
    * called.
    * <p>
    * Completed RSS feed loads can be retrieved by calling {@link #take()}.
    * Alternatively, non-blocking polling is possible with {@link #poll()}.
    * 
    * @param uri
    *          RSS feed URI to be loaded
    * @param priority
    *          larger integer gives higher priority
    * 
    * @return Future representing the RSS feed scheduled for loading,
    *         {@code null} if scheduling failed
    */
   public Future<RSSFeed> load(String uri, int priority) {
     if (uri == null) {
       throw new IllegalArgumentException("RSS feed URI must not be null.");
     }
 
     // optimization (after flag changes have become visible)
     if (stopped) {
       return null;
     }
 
     // flag readings happen-after enqueue
     final RSSFuture future = new RSSFuture(uri, priority);
     final boolean ok = in.offer(future);
 
     if (!ok || stopped) {
       return null;
     }
 
     return future;
   }
 
   /**
    * Retrieves and removes the next Future representing the result of loading an
    * RSS feed, waiting if none are yet present.
    * 
    * @return the {@link Future} representing the loaded RSS feed
    * 
    * @throws InterruptedException
    *           if interrupted while waiting
    */
   public Future<RSSFeed> take() throws InterruptedException {
     return out.take();
   }
 
   /**
    * Retrieves and removes the next Future representing the result of loading an
    * RSS feed or {@code null} if none are present.
    * 
    * @return the {@link Future} representing the loaded RSS feed, or
    *         {@code null} if none are present
    * 
    * @throws InterruptedException
    *           if interrupted while waiting
    */
   public Future<RSSFeed> poll() {
     return out.poll();
   }
 
   /**
    * Retrieves and removes the Future representing the result of loading an RSS
    * feed, waiting if necessary up to the specified wait time if none are yet
    * present.
    * 
    * @param timeout
    *          how long to wait before giving up, in units of {@code unit}
    * @param unit
    *          a {@link TimeUnit} determining how to interpret the
    *          {@code timeout} parameter
    * @return the {@link Future} representing the loaded RSS feed, or
    *         {@code null} if none are present within the specified time interval
    * @throws InterruptedException
    *           if interrupted while waiting
    */
   public Future<RSSFeed> poll(long timeout, TimeUnit unit) throws InterruptedException {
     return out.poll(timeout, unit);
   }
 
   /**
    * Internal consumer of RSS feed URIs stored in the blocking queue.
    */
   class Loader implements Runnable {
 
     private final RSSReader reader;
 
     Loader(RSSReader reader) {
       this.reader = reader;
     }
 
     /**
      * Keep on loading RSS feeds by dequeuing incoming tasks until the sentinel
      * is encountered.
      */
     @Override
     public void run() {
       try {
         RSSFuture future = null;
         RSSFeed feed;
         while ((future = in.take()) != SENTINEL) {
 
           if (future.status.compareAndSet(RSSFuture.READY, RSSFuture.LOADING)) {
             try {
               // perform loading outside of locked region
               feed = reader.load(future.uri);
 
               // set successfully loaded RSS feed
               future.set(feed, /* error */null);
 
               // enable caller to consume the loaded RSS feed
               out.add(future);
             } catch (RSSException e) {
               // throw ExecutionException when calling RSSFuture::get()
               future.set(/* feed */null, e);
             } catch (RSSFault e) {
               // throw ExecutionException when calling RSSFuture::get()
               future.set(/* feed */null, e);
             } finally {
               // RSSFuture::isDone() returns true even if an error occurred
               future.status.compareAndSet(RSSFuture.LOADING, RSSFuture.LOADED);
             }
           }
 
         }
       } catch (InterruptedException e) {
         // Restore the interrupted status
         Thread.currentThread().interrupt();
       }
     }
 
   }
 
   /**
    * Internal sentinel to stop the thread that is loading RSS feeds.
    */
   private final static RSSFuture SENTINEL = new RSSFuture(null, /* priority */7);
 
   /**
    * Offer callers control over the asynchronous loading of an RSS feed.
    */
   static class RSSFuture implements Future<RSSFeed>, Comparable<RSSFuture> {
 
     static final int DEFAULT_PRIORITY = 3;
     static final int READY = 0;
     static final int LOADING = 1;
     static final int LOADED = 2;
     static final int CANCELLED = 4;
 
     /** RSS feed URI */
     final String uri;
 
     /** Larger integer gives higher priority */
     final int priority;
 
     AtomicInteger status;
 
     boolean waiting;
     RSSFeed feed;
     Exception cause;
 
     RSSFuture(String uri, int priority) {
       this.uri = uri;
       this.priority = priority;
       status = new AtomicInteger(READY);
     }
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
       return isCancelled() || status.compareAndSet(READY, CANCELLED);
     }
 
     @Override
     public boolean isCancelled() {
       return status.get() == CANCELLED;
     }
 
     @Override
     public boolean isDone() {
       return (status.get() & (LOADED | CANCELLED)) != 0;
     }
 
     @Override
     public synchronized RSSFeed get() throws InterruptedException, ExecutionException {
       if (feed == null && cause == null) {
         try {
           waiting = true;
 
           // guard against spurious wakeups
           while (waiting) {
             wait();
           }
         } finally {
           waiting = false;
         }
       }
 
       if (cause != null) {
         throw new ExecutionException(cause);
       }
 
       return feed;
     }
 
     @Override
     public synchronized RSSFeed get(long timeout, TimeUnit unit)
         throws InterruptedException, ExecutionException, TimeoutException {
 
       if (feed == null && cause == null) {
         try {
           waiting = true;
 
           final long timeoutMillis = unit.toMillis(timeout);
           final long startMillis = System.currentTimeMillis();
 
           // guard against spurious wakeups
           while (waiting) {
             wait(timeoutMillis);
 
             // check timeout
             if (System.currentTimeMillis() - startMillis > timeoutMillis) {
               throw new TimeoutException("RSS feed loading timed out");
             }
           }
         } finally {
           waiting = false;
         }
       }
 
       if (cause != null) {
         throw new ExecutionException(cause);
       }
 
       return feed;
     }
 
     synchronized void set(RSSFeed feed, Exception cause) {
       this.feed = feed;
       this.cause = cause;
 
       if (waiting) {
         waiting = false;
         notifyAll();
       }
     }
 
     @Override
     public int compareTo(RSSFuture other) {
       // Note: head of PriorityQueue implementation is the least element
       return other.priority - priority;
     }
   }
 
 }