tmf: Make HistoryBuilder use a TMF request for its events

Initially, the IStateChangeInput object would have to open
the trace itself, and merely send state changes. Rework this
interface so that we now have to pass it the events. This
allows HistoryBuilder to hook into TMF's event requests, and
benefit from the coalescing.

This will be tweaked a bit more soon, when we put all indexing
requests together.

Signed-off-by: Alexandre Montplaisir <alexmonthy@voxpopuli.im>

2 files changed, 42 insertions, 36 deletions

diff --git a/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/internal/lttng2/kernel/core/stateprovider/CtfKernelStateInput.java b/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/internal/lttng2/kernel/core/stateprovider/CtfKernelStateInput.java
index 4470b8a767..392124d93d 100644
--- a/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/internal/lttng2/kernel/core/stateprovider/CtfKernelStateInput.java
+++ b/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/internal/lttng2/kernel/core/stateprovider/CtfKernelStateInput.java
@@ -16,9 +16,9 @@ import java.io.IOException;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
-import org.eclipse.linuxtools.tmf.core.ctfadaptor.CtfIterator;
 import org.eclipse.linuxtools.tmf.core.ctfadaptor.CtfTmfEvent;
 import org.eclipse.linuxtools.tmf.core.ctfadaptor.CtfTmfTrace;
+import org.eclipse.linuxtools.tmf.core.event.ITmfEvent;
 import org.eclipse.linuxtools.tmf.core.statesystem.IStateChangeInput;
 import org.eclipse.linuxtools.tmf.core.statesystem.IStateSystemBuilder;
 
@@ -33,11 +33,11 @@ import org.eclipse.linuxtools.tmf.core.statesystem.IStateSystemBuilder;
  */
 public class CtfKernelStateInput implements IStateChangeInput {
 
-    final static int EVENTS_QUEUE_SIZE = 10000;
+    private static final int EVENTS_QUEUE_SIZE = 10000;
 
     private final BlockingQueue<CtfTmfEvent> eventsQueue;
 
-    private final CtfIterator traceReader;
+    private final CtfTmfTrace trace;
     private final CtfKernelHandler eventHandler;
 
     private final Thread eventHandlerThread;
@@ -55,7 +55,7 @@ public class CtfKernelStateInput implements IStateChangeInput {
      */
     public CtfKernelStateInput(CtfTmfTrace trace) {
         eventsQueue = new ArrayBlockingQueue<CtfTmfEvent>(EVENTS_QUEUE_SIZE);
-        traceReader = new CtfIterator(trace);
+        this.trace = trace;
         eventHandler = new CtfKernelHandler(eventsQueue);
         ssAssigned = false;
 
@@ -64,43 +64,54 @@ public class CtfKernelStateInput implements IStateChangeInput {
     }
 
     @Override
-    public void run() {
-        if (!ssAssigned) {
-            System.err.println("Cannot start Input thread without a target state system"); //$NON-NLS-1$
-            return;
-        }
-
-        CtfTmfEvent currentEvent;
-
-        eventHandlerThread.start();
+    public CtfTmfTrace getTrace() {
+        return trace;
+    }
 
-        try {
-            currentEvent = traceReader.getCurrentEvent();
-            while (currentEvent != null) {
-                traceReader.advance();
-                eventsQueue.put(currentEvent);
-                currentEvent = traceReader.getCurrentEvent();
-            }
-            /*
-             * We're done reading the trace, insert a null event in the queue to
-             * stop the handler
-             */
-            eventsQueue.put(CtfTmfEvent.getNullEvent());
-            eventHandlerThread.join();
+    @Override
+    public long getStartTime() {
+        return trace.getStartTime().getValue();
+    }
 
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
+    @Override
+    public CtfTmfEvent getExpectedEventType() {
+        return CtfTmfEvent.getNullEvent();
     }
 
     @Override
     public void assignTargetStateSystem(IStateSystemBuilder ssb) {
         eventHandler.assignStateSystem(ssb);
         ssAssigned = true;
+        eventHandlerThread.start();
     }
 
     @Override
-    public long getStartTime() {
-        return traceReader.getCtfTmfTrace().getStartTime().getValue();
+    public void processEvent(ITmfEvent event) {
+        /* Make sure the target state system has been assigned */
+        if (!ssAssigned) {
+            System.err.println("Cannot process event without a target state system"); //$NON-NLS-1$
+            return;
+        }
+        
+        /* Insert the event we're received into the events queue */
+        CtfTmfEvent currentEvent = (CtfTmfEvent) event;
+        try {
+            eventsQueue.put(currentEvent);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Override
+    public void dispose() {
+        /* Insert a null event in the queue to stop the event handler's thread. */
+        try {
+            eventsQueue.put(CtfTmfEvent.getNullEvent());
+            eventHandlerThread.join();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        ssAssigned = false;
+        eventHandler.assignStateSystem(null);
     }
 }
diff --git a/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/lttng2/kernel/core/trace/CtfKernelTrace.java b/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/lttng2/kernel/core/trace/CtfKernelTrace.java
index e49499b48d..bbbffcb47b 100644
--- a/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/lttng2/kernel/core/trace/CtfKernelTrace.java
+++ b/lttng/org.eclipse.linuxtools.lttng2.kernel.core/src/org/eclipse/linuxtools/lttng2/kernel/core/trace/CtfKernelTrace.java
@@ -95,10 +95,5 @@ public class CtfKernelTrace extends CtfTmfTrace {
 
         this.ss = builder.getStateSystemQuerier();
         builder.run(); /* Start the construction of the history */
-
-        //FIXME We will have to call close() once we are notified that the
-        //construction is done. Until this is implemented, we will just
-        //block here.
-        builder.close();
     }
 }