Handle count option on readers when streaming.

pdal/Reader.hpp

@@ -54,6 +54,8 @@ class PDAL_DLL Reader : public virtual Stage
 
     void setReadCb(PointReadFunc cb)
         { m_cb = cb; }
+    point_count_t count() const
+        { return m_count; }
 
 protected:
     std::string m_filename;

pdal/Streamable.cpp

@@ -35,6 +35,7 @@
 #include <iterator>
 
 #include <pdal/Streamable.hpp>
+#include <pdal/Reader.hpp>
 
 namespace pdal
 {
@@ -169,6 +170,11 @@ void Streamable::execute(StreamPointTable& table,
     // Separate out the first stage.
     Streamable *reader = stages.front();
 
+    // We may be limited in the number of points requested.
+    point_count_t count = std::numeric_limits<point_count_t>::max();
+    if (Reader *r = dynamic_cast<Reader *>(reader))
+        count = r->count();
+
     // Build a list of all stages except the first.  We may have a writer in
     // this list in addition to filters, but we treat them in the same way.
     auto begin = stages.begin();
@@ -185,7 +191,7 @@ void Streamable::execute(StreamPointTable& table,
         table.clearSpatialReferences();
         PointId idx = 0;
         PointRef point(table, idx);
-        point_count_t pointLimit = table.capacity();
+        point_count_t pointLimit = std::min(count, table.capacity());
 
         reader->startLogging();
         // When we get false back from a reader, we're done, so set
@@ -201,6 +207,8 @@ void Streamable::execute(StreamPointTable& table,
             if (finished)
                 pointLimit = idx;
         }
+        count -= pointLimit;
+
         reader->stopLogging();
         srs = reader->getSpatialReference();
         if (!srs.empty())

test/unit/StreamingTest.cpp

@@ -399,3 +399,37 @@ TEST(Streaming, issue_2069)
     t.execute(table);
 }
 
+// Make sure that we take the "count" option into account when streaming.
+TEST(Streaming, issue_2086)
+{
+    StageFactory f;
+    Stage *r = f.createStage("readers.las");
+    Options opts;
+    opts.add("filename", Support::datapath("las/autzen_trim.las"));
+    opts.add("count", 35U);
+    r->setOptions(opts);
+
+    StreamCallbackFilter streamFilter;
+
+    int cnt = 0;
+    auto cb = [&cnt](PointRef&)
+    {
+        cnt++;
+        return true;
+    };
+    streamFilter.setCallback(cb);
+    streamFilter.setInput(*r);
+
+    // Use a table size smaller than count.
+    FixedPointTable t(20);
+    streamFilter.prepare(t);
+    streamFilter.execute(t);
+    EXPECT_EQ(cnt, 35);
+
+    cnt = 0;
+    // Use a table size larger than count.
+    FixedPointTable t2(100);
+    streamFilter.prepare(t2);
+    streamFilter.execute(t2);
+    EXPECT_EQ(cnt, 35);
+}