1 files changed, 19 insertions, 29 deletions

diff --git a/src/de/lmu/ifi/dbs/elki/datasource/InputStreamDatabaseConnection.java b/src/de/lmu/ifi/dbs/elki/datasource/InputStreamDatabaseConnection.java
index 0e99b421..21439807 100644
--- a/src/de/lmu/ifi/dbs/elki/datasource/InputStreamDatabaseConnection.java
+++ b/src/de/lmu/ifi/dbs/elki/datasource/InputStreamDatabaseConnection.java
@@ -4,7 +4,7 @@ package de.lmu.ifi.dbs.elki.datasource;
  This file is part of ELKI:
  Environment for Developing KDD-Applications Supported by Index-Structures
 
- Copyright (C) 2013
+ Copyright (C) 2014
  Ludwig-Maximilians-Universität München
  Lehr- und Forschungseinheit für Datenbanksysteme
  ELKI Development Team
@@ -38,8 +38,7 @@ import de.lmu.ifi.dbs.elki.utilities.documentation.Title;
 import de.lmu.ifi.dbs.elki.utilities.optionhandling.parameterization.Parameterization;
 
 /**
- * Provides a database connection expecting input from an input stream such as
- * stdin.
+ * Database connection expecting input from an input stream such as stdin.
  * 
  * @author Arthur Zimek
  * 
@@ -80,48 +79,39 @@ public class InputStreamDatabaseConnection extends AbstractDatabaseConnection {
     if(LOG.isDebugging()) {
       LOG.debugFine("Invoking parsers.");
     }
+    // Streaming parsers may yield to stream filters immediately.
     if(parser instanceof StreamingParser) {
-      final StreamingParser streamParser = (StreamingParser)parser;
+      final StreamingParser streamParser = (StreamingParser) parser;
       streamParser.initStream(in);
-
       // normalize objects and transform labels
       if(LOG.isDebugging()) {
-        LOG.debugFine("Invoking filters.");
-      }
-      Duration duration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".load") : null;
-      if (duration != null) {
-        duration.begin();
+        LOG.debugFine("Parsing as stream.");
       }
-      MultipleObjectsBundle objects = MultipleObjectsBundle.fromStream(invokeFilters(streamParser));
-      if (duration != null) {
-        duration.end();
-        LOG.statistics(duration);
+      Duration duration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".load").begin() : null;
+      MultipleObjectsBundle objects = invokeStreamFilters(streamParser).asMultipleObjectsBundle();
+      parser.cleanup();
+      if(duration != null) {
+        LOG.statistics(duration.end());
       }
       return objects;
     }
     else {
-      Duration duration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".parse") : null;
-      if (duration != null) {
-        duration.begin();
-      }
+      // For non-streaming parsers, we first parse, then filter
+      Duration duration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".parse").begin() : null;
       MultipleObjectsBundle parsingResult = parser.parse(in);
-      if (duration != null) {
-        duration.end();
-        LOG.statistics(duration);
+      parser.cleanup();
+      if(duration != null) {
+        LOG.statistics(duration.end());
       }
 
       // normalize objects and transform labels
       if(LOG.isDebugging()) {
         LOG.debugFine("Invoking filters.");
       }
-      Duration fduration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".filter") : null;
-      if (fduration != null) {
-        fduration.begin();
-      }
-      MultipleObjectsBundle objects = invokeFilters(parsingResult);
-      if (fduration != null) {
-        fduration.end();
-        LOG.statistics(fduration);
+      Duration fduration = LOG.isStatistics() ? LOG.newDuration(this.getClass().getName() + ".filter").begin() : null;
+      MultipleObjectsBundle objects = invokeBundleFilters(parsingResult);
+      if(fduration != null) {
+        LOG.statistics(fduration.end());
       }
       return objects;
     }