org.eclipse.linuxtools.tmf/src/org/eclipse/linuxtools/tmf/request/TmfDataRequest.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009 Ericsson
   3  *
   4  * All rights reserved. This program and the accompanying materials are
   5  * made available under the terms of the Eclipse Public License v1.0 which
   6  * accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *   Francois Chouinard - Initial API and implementation
  11  *******************************************************************************/
  12
  13 package org.eclipse.linuxtools.tmf.request;
  14
  15 import org.eclipse.linuxtools.tmf.event.TmfTimeRange;
  16
  17 /**
  18  * <b><u>TmfDataRequest</u></b>
  19  * <p>
  20  * TmfDataRequests are used to obtain blocks of contiguous data from a data
  21  * provider, either all the data within a given time window or n elements
  22  * starting at a specific timestamp. Open ranges can be used, especially for
  23  * continuous streaming.
  24  * <p>
  25  * The request is processed asynchronously by an ITmfRequestProcessor and,
  26  * as blocks of data become available, the callback handleData() is
  27  * invoked, synchronously, for each block. When returning from the callback,
  28  * the data instances go out of scope and become eligible for gc. It is
  29  * is thus the responsibility of the requester to either copy or keep a
  30  * reference to the data it wishes to track specifically.
  31  * <p>
  32  * This data block approach is necessary to avoid  busting the heap for very
  33  * large trace files. The block size is configurable.
  34  * <p>
  35  * The ITmfRequestProcessor indicates that the request is completed by
  36  * calling done(). The request can be canceled at any time with cancel().
  37  * <p>
  38  * Typical usage:
  39  *<pre><code><i>TmfTimeWindow range = new TmfTimewindow(...);
  40  *TmfDataRequest&lt;DataType[]&gt; request = new TmfDataRequest&lt;DataType[]&gt;(range, 0, NB_EVENTS, BLOCK_SIZE) {
  41  *    public void handleData() {
  42  *         DataType[] data = request.getData();
  43  *         for (DataType e : data) {
  44  *             // do something
  45  *         }
  46  *    }
  47  *    public void handleSuccess() {
  48  *         // do something
  49  *         }
  50  *    }
  51  *    public void handleFailure() {
  52  *         // do something
  53  *         }
  54  *    }
  55  *    public void handleCancel() {
  56  *         // do something
  57  *         }
  58  *    }
  59  *};
  60  *fProcessor.process(request, true);
  61  *</i></code></pre>
  62  *
  63  * TODO: Consider extending DataRequestMonitor from DSF concurrency plugin.
  64  * The main issue is the slicing of the result in blocks and continuous
  65  * streams. This would require using a thread executor and to carefully
  66  * look at setData() and getData().
  67  *
  68  * TODO: Implement request failures (codes, etc...)
  69  */
  70 public class TmfDataRequest<V> {
  71
  72     // ========================================================================
  73     // Constants
  74     // ========================================================================
  75
  76     // The default maximum number of events per chunk
  77     public static final int DEFAULT_BLOCK_SIZE = 1000;
  78
  79     // The request count for all the events
  80     public static final int ALL_EVENTS = -1;
  81
  82     // ========================================================================
  83     // Attributes
  84     // ========================================================================
  85
  86     private final int  fIndex;              // The index (order) of the requested event
  87     private final TmfTimeRange fRange;      // The requested events time range
  88     private final int  fNbRequestedEvents;  // The number of events to read (-1 == all in the range)
  89     private final int  fBlockSize;          // The maximum number of events per chunk
  90     private       int  fNbEvents;           // The number of events read so far
  91
  92     private Object lock = new Object();
  93     private boolean fRequestCompleted = false;
  94     private boolean fRequestFailed    = false;
  95     private boolean fRequestCanceled  = false;
  96
  97     private V[] fData;  // Data object
  98
  99     // ========================================================================
 100     // Constructors
 101     // ========================================================================
 102
 103     /**
 104      * @param index
 105      * @param nbEvents
 106      */
 107     public TmfDataRequest(int index, int nbEvents) {
 108         this(null, index, nbEvents, DEFAULT_BLOCK_SIZE);
 109     }
 110
 111     public TmfDataRequest(int index, int nbEvents, int blockSize) {
 112         this(null, index, nbEvents, blockSize);
 113     }
 114
 115     /**
 116      * @param range
 117      */
 118     public TmfDataRequest(TmfTimeRange range) {
 119         this(range, 0, ALL_EVENTS, DEFAULT_BLOCK_SIZE);
 120     }
 121
 122     /**
 123      * @param range
 124      * @param nbEvents
 125      */
 126     public TmfDataRequest(TmfTimeRange range, int nbEvents) {
 127         this(range, 0, nbEvents, DEFAULT_BLOCK_SIZE);
 128     }
 129     /**
 130      * @param range
 131      * @param nbEvents
 132      * @param blockSize Size of the largest blocks expected
 133      */
 134     public TmfDataRequest(TmfTimeRange range, int nbEvents, int blockSize) {
 135         this(range, 0, nbEvents, blockSize);
 136     }
 137
 138     /**
 139      * @param range
 140      * @param index
 141      * @param nbEvents
 142      * @param blockSize Size of the largest blocks expected
 143      */
 144     private TmfDataRequest(TmfTimeRange range, int index, int nbEvents, int blockSize) {
 145         fIndex             = index;
 146         fRange             = range;
 147         fNbRequestedEvents = nbEvents;
 148         fBlockSize         = blockSize;
 149         fNbEvents          = 0;
 150     }
 151
 152     // ========================================================================
 153     // Accessors
 154     // ========================================================================
 155
 156         /**
 157          * @return the index
 158          */
 159         public int getIndex() {
 160                 return fIndex;
 161         }
 162
 163     /**
 164      * @return the requested time range
 165      */
 166     public TmfTimeRange getRange() {
 167         return fRange;
 168     }
 169
 170     /**
 171      * @return the number of requested events (-1 = all)
 172      */
 173     public int getNbRequestedEvents() {
 174         return fNbRequestedEvents;
 175     }
 176
 177     /**
 178      * @return the block size
 179      */
 180     public int getBlockize() {
 181         return fBlockSize;
 182     }
 183
 184     /**
 185      * @return the number of events read so far
 186      */
 187     public int getNbEvents() {
 188         return fNbEvents;
 189     }
 190
 191     /**
 192      * @return indicates if the request is completed
 193      */
 194     public boolean isCompleted() {
 195         return fRequestCompleted;
 196     }
 197
 198     /**
 199      * @return indicates if the request is canceled
 200      */
 201     public boolean isFailed() {
 202         return fRequestFailed;
 203     }
 204
 205     /**
 206      * @return indicates if the request is canceled
 207      */
 208     public boolean isCancelled() {
 209         return fRequestCanceled;
 210     }
 211
 212     // ========================================================================
 213     // Operators
 214     // ========================================================================
 215
 216     /**
 217      * Sets the data object to specified value. To be called by the
 218      * asynchronous method implementor.
 219      * @param data Data value to set.
 220      */
 221     public synchronized void setData(V[] data) {
 222         fNbEvents += data.length;
 223         fData = data;
 224     }
 225
 226     /**
 227      * Returns the data value, null if not set.
 228      */
 229     public synchronized V[] getData() {
 230         return fData;
 231     }
 232
 233     /**
 234      * Handle a block of incoming data. This method is called every time
 235      * a block of data becomes available.
 236      *
 237      * - Data items are received in the order they appear in the stream.
 238      * - Called by the request processor, in its execution thread, every time a
 239      *   block of data becomes available.
 240      * - Request processor performs a synchronous call to handlePartialResult()
 241      *   i.e. its execution threads holds until handlePartialData() returns.
 242      * - Original data items are disposed of on return i.e. keep a reference
 243      *   (or a copy) if some persistence is needed between invocations.
 244      * - When there is no more data, done() is called.
 245      *
 246      * @param events - an array of events
 247      */
 248     public void handleData() {
 249     }
 250
 251     /**
 252      * Handle the completion of the request. It is called when there is no more
 253      * data available either because:
 254      * - the request completed normally
 255      * - the request failed
 256      * - the request was canceled
 257      *
 258      * As a convenience, handleXXXX methods are provided. They are meant to be
 259      * overridden by the application if it needs to handle these conditions.
 260      */
 261     public void handleCompleted() {
 262         if (fRequestFailed) {
 263                 handleFailure();
 264         }
 265         else if (fRequestCanceled) {
 266                 handleCancel();
 267         }
 268         else {
 269                 handleSuccess();
 270         }
 271     }
 272
 273     public void handleSuccess() {
 274     }
 275
 276     public void handleFailure() {
 277     }
 278
 279     public void handleCancel() {
 280     }
 281
 282     /**
 283      * To suspend the client thread until the request completes (or is
 284      * canceled).
 285      *
 286      * @throws InterruptedException
 287      */
 288     public void waitForCompletion() {
 289         synchronized (lock) {
 290             while (!fRequestCompleted)
 291                 try {
 292                     lock.wait();
 293                 } catch (InterruptedException e) {
 294                     e.printStackTrace();
 295                 }
 296         }
 297     }
 298
 299     /**
 300      * Called by the request processor upon completion.
 301      */
 302     public void done() {
 303         synchronized(lock) {
 304             fRequestCompleted = true;
 305             lock.notify();
 306         }
 307         handleCompleted();
 308     }
 309
 310     /**
 311      * Called by the request processor upon failure.
 312      */
 313     public void fail() {
 314         synchronized(lock) {
 315             fRequestFailed = true;
 316             done();
 317         }
 318     }
 319
 320     /**
 321      * Called by the request processor upon cancellation.
 322      */
 323     public void cancel() {
 324         synchronized(lock) {
 325             fRequestCanceled = true;
 326             done();
 327         }
 328     }
 329
 330 }