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.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> eventType) {
  40         super.init(name, eventType);
  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>(fType, 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             }
  79             fPendingCoalescedRequests.add(coalescedRequest);
  80         } else {
  81             super.newCoalescedDataRequest(request);
  82         }
  83     }
  84
  85         @Override
  86         protected void queueBackgroundRequest(final ITmfDataRequest<T> request, final int blockSize, final boolean indexing) {
  87
  88                 if (! (request instanceof ITmfEventRequest)) {
  89                         super.queueBackgroundRequest(request, blockSize, indexing);
  90                         return;
  91                 }
  92
  93                 Thread thread = new Thread() {
  94                         @Override
  95                         public void run() {
  96                                 request.start();
  97
  98                                 final Integer[] CHUNK_SIZE = new Integer[1];
  99                                 CHUNK_SIZE[0] = Math.min(request.getNbRequested(), blockSize + ((indexing) ? 1 : 0));
 100
 101                                 final Integer[] nbRead = new Integer[1];
 102                                 nbRead[0] = 0;
 103
 104                                 final Boolean[] isFinished = new Boolean[1];
 105                                 isFinished[0] = Boolean.FALSE;
 106
 107                                 int startIndex = request.getIndex();
 108
 109                                 while (!isFinished[0]) {
 110
 111                                         TmfEventRequest<T> subRequest= new TmfEventRequest<T>(request.getDataType(), ((ITmfEventRequest<?>) request).getRange(), startIndex + nbRead[0], CHUNK_SIZE[0], blockSize, ExecutionType.BACKGROUND)
 112                                         {
 113                                                 @Override
 114                                                 public void handleData(T data) {
 115                                                         super.handleData(data);
 116                                                         if (request.getDataType().isInstance(data)) {
 117                                                             request.handleData(data);
 118                                                         }
 119                                                         if (this.getNbRead() > CHUNK_SIZE[0]) {
 120                                                                 System.out.println("ERROR - Read too many events"); //$NON-NLS-1$
 121                                                         }
 122                                                 }
 123
 124                                                 @Override
 125                                                 public void handleCompleted() {
 126                                                         nbRead[0] += this.getNbRead();
 127                                                         if (nbRead[0] >= request.getNbRequested() || (this.getNbRead() < CHUNK_SIZE[0])) {
 128                                                                 if (this.isCancelled()) {
 129                                                                         request.cancel();
 130                                                                 } else if (this.isFailed()) {
 131                                                                     request.fail();
 132                                                                 } else {
 133                                                                         request.done();
 134                                                                 }
 135                                                                 isFinished[0] = Boolean.TRUE;
 136                                                         }
 137                                                         super.handleCompleted();
 138                                                 }
 139                                         };
 140
 141                                         if (!isFinished[0]) {
 142                                                 queueRequest(subRequest);
 143
 144                                                 try {
 145                                                         subRequest.waitForCompletion();
 146                                                 } catch (InterruptedException e) {
 147                                                         e.printStackTrace();
 148                                                 }
 149
 150                                                 if (startIndex == 0 && nbRead[0].equals(CHUNK_SIZE[0])) { // do this only once if the event request index is unknown
 151                                                         startIndex = subRequest.getIndex(); // update the start index with the index of the first subrequest's
 152                                                 }                                       // start time event which was set during the arm request
 153                                                 CHUNK_SIZE[0] = Math.min(request.getNbRequested() - nbRead[0], blockSize);
 154                                         }
 155                                 }
 156                         }
 157                 };
 158
 159                 thread.start();
 160         }
 161 }