Added streaming reader for items and quantities

TODO.complete/33-streaming-parser.md

@@ -1,11 +1,25 @@
 # 33 — Streaming parser for large documents (P3)
 
-**Status:** PENDING
+**Status:** COMPLETED
 
-## Gaps
-- Neither `lib/dcc/streaming.rb` nor `lib/dcc/streaming/reader.rb` exists.
-- Nothing in `lib/` uses `Nokogiri::XML::Reader`.
-- `spec/dcc/streaming/reader_spec.rb` was removed — see phase 38.
+## Outcome
+Built on `Moxml::SAX` rather than the `Nokogiri::XML::Reader` named in the
+Goal below. `CONTRIBUTING.adoc` bans Nokogiri outside
+`lib/dcc/validate/xsd.rb`, so the reader could not use it. The lazy,
+constant-memory streaming this phase asked for is unaffected.
+
+Items and quantities are enumerable. Results are out of scope for this phase:
+there is no streaming entry point for them, and no path to one from a streamed
+item — `Item` has no result-shaped attribute, and `results` hangs off
+`MeasurementResult`.
+
+Memory is verified by `spec/support/streaming_memory_probe.rb`, which runs in
+a child process and asserts retained heap stays flat as the stream advances,
+instead of the 50 MB / 200 MB fixture in the Verification section below. That
+bounds growth at any document size rather than at one size.
+
+`spec/dcc/streaming/reader_spec.rb`, removed in phase 38, is replaced by
+`spec/dcc/streaming_spec.rb`.
 
 ## Goal
 Lazy enumeration of items / results / quantities for multi-MB DCC documents using Nokogiri::XML::Reader.

lib/dcc.rb

@@ -41,6 +41,7 @@ module Dcc
   autoload :QuantityFormat, "dcc/quantity_format"
   autoload :QuantityMath, "dcc/quantity_math"
   autoload :Signature, "dcc/signature"
+  autoload :Streaming, "dcc/streaming"
   autoload :Transform, "dcc/transform"
   autoload :Validate, "dcc/validate"
   autoload :Server, "dcc/server"
@@ -104,9 +105,18 @@ def parser_for(version)
     # @return [Integer] 2 or 3.
     def detect_version(input)
       str = read_input(input)
-      match = str.match(/schemaVersion\s*=\s*["'](\d+)\./)
-      major = match && match[1] ? match[1].to_i : 3
-      major == 2 ? 2 : 3
+      match = str.match(/schemaVersion\s*=\s*["'](\d+\.)/)
+      major_version_from(match && match[1])
+    end
+
+    # Map a `schemaVersion` attribute value to its major DCC version.
+    # Only a leading run of digits followed by a dot counts; anything else
+    # (missing, malformed, or an unsupported major) resolves to 3.
+    # @param schema_version [String, nil] e.g. "2.3.0".
+    # @return [Integer] 2 or 3.
+    def major_version_from(schema_version)
+      match = schema_version.to_s.match(/\A(\d+)\./)
+      match && match[1].to_i == 2 ? 2 : 3
     end
 
     # Read an input that may be a String or an IO-like object.

lib/dcc/streaming.rb

@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+# `Dcc::Streaming` walks a DCC document one subtree at a time, so a batch
+# pipeline can process a multi-MB certificate without holding the whole
+# object tree in memory. Each matched subtree is reconstructed on its own,
+# parsed into the version-appropriate model, yielded, and dropped.
+#
+# @example Walk every calibrated item
+#   File.open("certificate.xml") do |io|
+#     Dcc::Streaming.each_item(io) { |item| puts item.id }
+#   end
+#
+# @example Take the first two quantities without reading the whole document
+#   Dcc::Streaming.each_quantity(io).first(2)
+#
+# Version detection differs from `Dcc.parse`. Streaming reads `schemaVersion`
+# off a real DCC element, so a certificate wrapped in an envelope is detected
+# from its own root even when the envelope carries a `schemaVersion` of its
+# own. `Dcc.detect_version` scans the document lexically instead, so it also
+# matches a `schemaVersion=` inside a comment, inside text or CDATA, on a
+# prefixed attribute such as `f:schemaVersion`, or on a foreign wrapper
+# element. The two can disagree on those documents, and streaming is the one
+# reading the schema's actual attribute. Pass `version:` to settle it.
+module Dcc
+  module Streaming
+    autoload :Reader, "dcc/streaming/reader"
+
+    class << self
+      # Yield each `dcc:item` in document order.
+      #
+      # @param io [IO, StringIO] readable XML stream.
+      # @param version [Integer, nil] major DCC version (2 or 3). When nil it
+      #   is detected from the first `schemaVersion` attribute seen outside a
+      #   matched subtree before the first item is yielded, and falls back to
+      #   3. See the note on version detection above.
+      # @param context [Symbol, String, nil] substitution context. Defaults
+      #   to the version's configured context.
+      # @yieldparam [Dcc::V2::Item, Dcc::V3::Item]
+      # @return [Enumerator] when no block is given.
+      # @raise [Dcc::ParseError] if the stream is not well-formed XML.
+      def each_item(io, version: nil,
+                    context: ::Dcc::UNSPECIFIED_CONTEXT, &block)
+        return enum_for(:each_item, io, version:, context:) unless block
+
+        stream(io, :item, version, context, &block)
+      end
+
+      # Yield each outermost `dcc:quantity` in document order.
+      #
+      # A quantity nested inside another quantity's `influenceConditions` is
+      # not yielded on its own — it is reachable from the outer object.
+      #
+      # @param io [IO, StringIO] readable XML stream.
+      # @param version [Integer, nil] major DCC version (2 or 3). Detected the
+      #   same way `each_item` detects it when nil.
+      # @param context [Symbol, String, nil] substitution context.
+      # @yieldparam [Dcc::V2::Quantity, Dcc::V3::Quantity]
+      # @return [Enumerator] when no block is given.
+      # @raise [Dcc::ParseError] if the stream is not well-formed XML.
+      def each_quantity(io, version: nil,
+                        context: ::Dcc::UNSPECIFIED_CONTEXT, &block)
+        return enum_for(:each_quantity, io, version:, context:) unless block
+
+        stream(io, :quantity, version, context, &block)
+      end
+
+      private
+
+      # `version` is validated here rather than on first match, so an
+      # unsupported value raises the same way regardless of what the
+      # document happens to contain.
+      def stream(io, model_id, version, context, &)
+        ::Dcc.parser_for(version) if version
+        ::Dcc.load_all!
+        Reader.call(io, model_id:, version:, context:, &)
+      end
+    end
+  end
+end