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