Report actual non-streamable stage in error. (#2189)

* Report actual non-streamable stage in error.
Close #2186

* Rename function.

pdal/Stage.hpp

@@ -137,6 +137,16 @@ class PDAL_DLL Stage
     virtual bool pipelineStreamable() const
     { return false; }
 
+    /**
+      Return a pointer to a pipeline's first non-streamable stage,
+      if one exists.
+
+      \return  nullptr if the stage is streamable, a pointer to this stage
+        otherwise.
+    */
+    virtual const Stage *findNonstreamable() const
+    { return this; }
+
     /**
       Set the spatial reference of a stage.
 

pdal/Streamable.cpp

@@ -53,6 +53,20 @@ bool Streamable::pipelineStreamable() const
 }
 
 
+const Stage *Streamable::findNonstreamable() const
+{
+    const Stage *nonstreamable;
+
+    for (const Stage *s : m_inputs)
+    {
+        nonstreamable = s->findNonstreamable();
+        if (nonstreamable)
+            return nonstreamable;
+    }
+    return nullptr;
+}
+
+
 // Streamed execution.
 void Streamable::execute(StreamPointTable& table)
 {
@@ -100,9 +114,10 @@ void Streamable::execute(StreamPointTable& table)
         }
     };
 
-    if (!pipelineStreamable())
-        throwError("Attempting to use stream mode with a stage that doesn't "
-            "support streaming.");
+    const Stage *nonstreaming = findNonstreamable();
+    if (nonstreaming)
+        nonstreaming->throwError("Attempting to use stream mode with a "
+            "stage that doesn't support streaming.");
 
     SpatialReference srs;
     std::list<StreamableList> lists;

pdal/Streamable.hpp

@@ -107,6 +107,14 @@ class PDAL_DLL Streamable : public virtual Stage
     */
     virtual void spatialReferenceChanged(const SpatialReference& /*srs*/)
     {}
+
+    /**
+      Find the first nonstreamable stage in a pipeline.
+
+      \return  NULL if the pipeline is streamable, otherwise return
+        a pointer to the first found stage that's not streamable.
+    */
+    const Stage *findNonstreamable() const;
 };
 
 } // namespace pdal