1 files changed, 34 insertions, 68 deletions

diff --git a/src/de/lmu/ifi/dbs/elki/datasource/AbstractDatabaseConnection.java b/src/de/lmu/ifi/dbs/elki/datasource/AbstractDatabaseConnection.java
index 77cdb12c..9da9a550 100644
--- a/src/de/lmu/ifi/dbs/elki/datasource/AbstractDatabaseConnection.java
+++ b/src/de/lmu/ifi/dbs/elki/datasource/AbstractDatabaseConnection.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
@@ -27,7 +27,6 @@ import java.util.List;
 
 import de.lmu.ifi.dbs.elki.datasource.bundle.BundleStreamSource;
 import de.lmu.ifi.dbs.elki.datasource.bundle.MultipleObjectsBundle;
-import de.lmu.ifi.dbs.elki.datasource.bundle.StreamFromBundle;
 import de.lmu.ifi.dbs.elki.datasource.filter.ObjectFilter;
 import de.lmu.ifi.dbs.elki.datasource.filter.StreamFilter;
 import de.lmu.ifi.dbs.elki.datasource.parser.Parser;
@@ -73,83 +72,50 @@ public abstract class AbstractDatabaseConnection implements DatabaseConnection {
    * @param bundle the objects to process
    * @return processed objects
    */
-  protected MultipleObjectsBundle invokeFilters(MultipleObjectsBundle bundle) {
-    BundleStreamSource prevs = null;
-    MultipleObjectsBundle prevb = bundle;
-    if(filters != null) {
-      for(ObjectFilter filter : filters) {
-        if(filter instanceof StreamFilter) {
-          StreamFilter sfilter = (StreamFilter) filter;
-          if(prevs != null) {
-            sfilter.init(prevs);
-          }
-          else {
-            sfilter.init(new StreamFromBundle(prevb));
-          }
-          prevs = sfilter;
-          prevb = null;
-        }
-        else {
-          if(prevs != null) {
-            prevb = filter.filter(MultipleObjectsBundle.fromStream(prevs));
-            prevs = null;
-          }
-          else {
-            prevb = filter.filter(prevb);
-            prevs = null;
-          }
-        }
-      }
-    }
-    if(prevb != null) {
-      return prevb;
+  protected MultipleObjectsBundle invokeBundleFilters(MultipleObjectsBundle bundle) {
+    if(filters == null) {
+      return bundle;
     }
-    else {
-      return MultipleObjectsBundle.fromStream(prevs);
+    // We dynamically switch between streaming and bundle operations.
+    BundleStreamSource stream = null;
+    for(ObjectFilter filter : filters) {
+      if(filter instanceof StreamFilter) {
+        StreamFilter sfilter = (StreamFilter) filter;
+        stream = sfilter.init((stream != null) ? stream : bundle.asStream());
+        bundle = null; // No longer a bundle
+      }
+      else {
+        bundle = filter.filter((bundle != null) ? bundle : stream.asMultipleObjectsBundle());
+        stream = null; // No longer a stream
+      }
     }
+    return (bundle != null) ? bundle : stream.asMultipleObjectsBundle();
   }
 
   /**
    * Transforms the specified list of objects and their labels into a list of
    * objects and their associations.
    * 
-   * @param bundle the objects to process
+   * @param stream the objects to process
    * @return processed objects
    */
-  protected BundleStreamSource invokeFilters(BundleStreamSource bundle) {
-    BundleStreamSource prevs = bundle;
-    MultipleObjectsBundle prevb = null;
-    if(filters != null) {
-      for(ObjectFilter filter : filters) {
-        if(filter instanceof StreamFilter) {
-          StreamFilter sfilter = (StreamFilter) filter;
-          if(prevs != null) {
-            sfilter.init(prevs);
-          }
-          else {
-            sfilter.init(new StreamFromBundle(prevb));
-          }
-          prevs = sfilter;
-          prevb = null;
-        }
-        else {
-          if(prevs != null) {
-            prevb = filter.filter(MultipleObjectsBundle.fromStream(prevs));
-            prevs = null;
-          }
-          else {
-            prevb = filter.filter(prevb);
-            prevs = null;
-          }
-        }
-      }
-    }
-    if(prevs != null) {
-      return prevs;
+  protected BundleStreamSource invokeStreamFilters(BundleStreamSource stream) {
+    if(filters == null) {
+      return stream;
     }
-    else {
-      return new StreamFromBundle(prevb);
+    // We dynamically switch between streaming and bundle operations.
+    MultipleObjectsBundle bundle = null;
+    for(ObjectFilter filter : filters) {
+      if(filter instanceof StreamFilter) {
+        stream = ((StreamFilter) filter).init((stream != null) ? stream : bundle.asStream());
+        bundle = null;
+      }
+      else {
+        bundle = filter.filter((bundle != null) ? bundle : stream.asMultipleObjectsBundle());
+        stream = null;
+      }
     }
+    return (stream != null) ? stream : bundle.asStream();
   }
 
   /**
@@ -187,7 +153,7 @@ public abstract class AbstractDatabaseConnection implements DatabaseConnection {
      * Filters
      */
     protected List<ObjectFilter> filters;
-    
+
     /**
      * Parser to use
      */