org.eclipse.linuxtools.tmf.core/src/org/eclipse/linuxtools/tmf/core/component/TmfEventProvider.java

   1 /*******************************************************************************
   2  * Copyright (c) 2009, 2010 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.core.component;
  14
  15 import org.eclipse.linuxtools.internal.tmf.core.Tracer;
  16 import org.eclipse.linuxtools.internal.tmf.core.request.TmfCoalescedEventRequest;
  17 import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
  18 import org.eclipse.linuxtools.tmf.core.event.ITmfTimestamp;
  19 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest;
  20 import org.eclipse.linuxtools.tmf.core.request.ITmfDataRequest.ExecutionType;
  21 import org.eclipse.linuxtools.tmf.core.request.ITmfEventRequest;
  22 import org.eclipse.linuxtools.tmf.core.request.TmfEventRequest;
  23
  24 /**
  25  * An extension of TmfDataProvider timestamped events providers.
  26  *
  27  * @version 1.0
  28  * @author Francois Chouinard
  29  */
  30 public abstract class TmfEventProvider<T extends ITmfEvent> extends TmfDataProvider<T> {
  31
  32     // ------------------------------------------------------------------------
  33     // Constructors
  34     // ------------------------------------------------------------------------
  35
  36     public TmfEventProvider() {
  37         super();
  38     }
  39
  40     @Override
  41     public void init(String name, Class<T> type) {
  42         super.init(name, type);
  43     }
  44
  45     public TmfEventProvider(String name, Class<T> type) {
  46         super(name, type);
  47     }
  48
  49     public TmfEventProvider(String name, Class<T> type, int queueSize) {
  50         super(name, type, queueSize);
  51     }
  52
  53     public TmfEventProvider(TmfEventProvider<T> other) {
  54         super(other);
  55     }
  56
  57     // ------------------------------------------------------------------------
  58     // TmfDataProvider
  59     // ------------------------------------------------------------------------
  60
  61     @Override
  62     public boolean isCompleted(ITmfDataRequest<T> request, T data, int nbRead) {
  63         boolean requestCompleted = super.isCompleted(request, data, nbRead);
  64         if (!requestCompleted && request instanceof ITmfEventRequest<?>) {
  65             ITmfTimestamp endTime = ((ITmfEventRequest<?>) request).getRange().getEndTime();
  66             return data.getTimestamp().compareTo(endTime, false) > 0;
  67         }
  68         return requestCompleted;
  69     }
  70
  71     @Override
  72     protected synchronized void newCoalescedDataRequest(ITmfDataRequest<T> request) {
  73         if (request instanceof ITmfEventRequest<?>) {
  74             ITmfEventRequest<T> eventRequest = (ITmfEventRequest<T>) request;
  75             TmfCoalescedEventRequest<T> coalescedRequest = new TmfCoalescedEventRequest<T>(eventRequest.getDataType(), eventRequest.getRange(),
  76                     eventRequest.getIndex(), eventRequest.getNbRequested(), eventRequest.getBlockSize(), eventRequest.getExecType());
  77             coalescedRequest.addRequest(eventRequest);
  78             if (Tracer.isRequestTraced()) {
  79                 Tracer.traceRequest(request, "COALESCED with " + coalescedRequest.getRequestId()); //$NON-NLS-1$
  80                 Tracer.traceRequest(coalescedRequest, "now contains " + coalescedRequest.getSubRequestIds()); //$NON-NLS-1$
  81             }
  82             fPendingCoalescedRequests.add(coalescedRequest);
  83         } else {
  84             super.newCoalescedDataRequest(request);
  85         }
  86     }
  87
  88         @Override
  89         protected void queueBackgroundRequest(final ITmfDataRequest<T> request, final int blockSize, final boolean indexing) {
  90
  91                 if (! (request instanceof ITmfEventRequest)) {
  92                         super.queueBackgroundRequest(request, blockSize, indexing);
  93                         return;
  94                 }
  95
  96         final TmfDataProvider<T> provider = this;
  97
  98                 Thread thread = new Thread() {
  99                         @Override
 100                         public void run() {
 101
 102                 if (Tracer.isRequestTraced()) {
 103                     Tracer.traceRequest(request, "is being serviced by " + provider.getName()); //$NON-NLS-1$
 104                 }
 105
 106                             request.start();
 107
 108                                 final Integer[] CHUNK_SIZE = new Integer[1];
 109                                 CHUNK_SIZE[0] = Math.min(request.getNbRequested(), blockSize + ((indexing) ? 1 : 0));
 110
 111                                 final Integer[] nbRead = new Integer[1];
 112                                 nbRead[0] = 0;
 113
 114                                 final Boolean[] isFinished = new Boolean[1];
 115                                 isFinished[0] = Boolean.FALSE;
 116
 117                                 long startIndex = request.getIndex();
 118
 119                                 while (!isFinished[0]) {
 120
 121                                         TmfEventRequest<T> subRequest= new TmfEventRequest<T>(request.getDataType(), ((ITmfEventRequest<?>) request).getRange(), startIndex + nbRead[0], CHUNK_SIZE[0], blockSize, ExecutionType.BACKGROUND)
 122                                         {
 123                                                 @Override
 124                                                 public void handleData(T data) {
 125                                                         super.handleData(data);
 126                                                         if (request.getDataType().isInstance(data)) {
 127                                                             request.handleData(data);
 128                                                         }
 129                                                         if (this.getNbRead() > CHUNK_SIZE[0]) {
 130                                                                 System.out.println("ERROR - Read too many events"); //$NON-NLS-1$
 131                                                         }
 132                                                 }
 133
 134                                                 @Override
 135                                                 public void handleCompleted() {
 136                                                         nbRead[0] += this.getNbRead();
 137                                                         if (nbRead[0] >= request.getNbRequested() || (this.getNbRead() < CHUNK_SIZE[0])) {
 138                                                                 if (this.isCancelled()) {
 139                                                                         request.cancel();
 140                                                                 } else if (this.isFailed()) {
 141                                                                     request.fail();
 142                                                                 } else {
 143                                                                         request.done();
 144                                                                 }
 145                                                                 isFinished[0] = Boolean.TRUE;
 146                                                         }
 147                                                         super.handleCompleted();
 148                                                 }
 149                                         };
 150
 151                                         if (!isFinished[0]) {
 152                                                 queueRequest(subRequest);
 153
 154                                                 try {
 155                                                         subRequest.waitForCompletion();
 156                                                 } catch (InterruptedException e) {
 157                                                         e.printStackTrace();
 158                                                 }
 159
 160                                                 if (startIndex == 0 && nbRead[0].equals(CHUNK_SIZE[0])) { // do this only once if the event request index is unknown
 161                                                         startIndex = subRequest.getIndex(); // update the start index with the index of the first subrequest's
 162                                                 }                                       // start time event which was set during the arm request
 163                                                 CHUNK_SIZE[0] = Math.min(request.getNbRequested() - nbRead[0], blockSize);
 164                                         }
 165                                 }
 166                         }
 167                 };
 168
 169                 thread.start();
 170         }
 171 }