The low-level changes to support this API I believe require us to implement the
bucket algorithm for the DataInputStream layer, so that the stream isn't being
read into an ever-growing buffer.
The key need/idea here is that the state, including our DaffodilInputStream
state, and it's current bitPos0b on the input stream, is preserved from one
parse call to the next so that the next parse picks up with the bitPos0b
exactly where the prior parse left it. That eliminates the java
InputStream/Channel issue - the caller provides that once, upon which it
disappears into the state.
This API I proposed thusfar provides no guarantees about how the java
InputStream is positioned after a parse call returns except that only a finite
section of it will have been consumed, meaning no seek to the end will have
been performed. The amount consumed may, as you observed, be more than what was
converted into the infoset, whether for pre-fetch reasons, or try-and-backtrack
reasons.
If one calls parse(), and a parse error prevents it from parsing anything, one
might want the InputStream positioned where it started. I think it may be
possible to provide this but only if the API requires a BufferedInputStream on
input. But I think we should leave this out, and require users to implement
this behavior themselves - something they can always do, by their own use of
BufferedInputStream's mark and reset calls. This repositioning back to where it
started may not be the desired behavior. Leaving the state such that yet
another parse call can be done.
I'd like to avoid exposing something as rich as DataInputStream to the API, but
I think making the State something that is passed to the dataProcessor
parse/unparse methods is probably essential.
If we allow the state object to be visible at the API, then in theory one could
pass it to different data processors that will then try to parse different
things from that stream/state. E.g., suppose you want to parse first a pcap
header, then you want to call parse() to fetch and parse each packet. You
cannot do that with the API as proposed.
This sounds like a real flaw in the design. There's lots of
header-body*-trailer data formats in the world where one does NOT want to
create a single document, but rather one wants to parse the header, then in a
loop parse each body record, and when that fails, parse the trailer record.
That allows one to process gigantic header-body*-trailer files in finite space.
So that's two things to reconsider:
1) precise InputStream behavior - providing some guarantees about where the
InputStream is positioned.
2) enabling different data processors to parse from the same InputStream if
called in sequence.
> I created a design note on Streaming API features we need for
> message-by-message processing style.
I think something that maybe isn't stated is missing, but I think is a
core justification for such a change. Thinking about it led me down a
brain dump:
My initial thought was that the new API looks very similar to something
like this:
val pf: ProcessorFactory = ???
val dp: DataProcessor = pf.onPath("/item")
val is: InputStream = ??? // the raw data
val xmlOut = new ScalaXMLInfosetOutputter()
def items : Stream[Node] = {
xmlOut.reset()
val pr = dp.parse(is, xmlOut)
val item = if (pr.isError) Nil else xmlOut.getResult()
item #:: items
}
Which is basically our existing API. And all we need to do is modify our
I/O layer to not be so greedy when it gets an InputStream like it is
now. But I think the main issue with this is that when we parse data we
could potentially read a bunch of data off the InputStream, then
backtrack, and now that InputStream data is lost for the next parse. To
me, that seems like a core issue with our current implementation that I
don't think you really mentioned.
To me, it seems the big thing that the StreamingParser gets you is that
I assume it would cache InputStream data from previous calls to parse()
so that they will be available to future calls of parse() if
backtracking occurs. Is this correct? Are the other benefits to the
StremaingParser?
If this is the main difference, rather than having a special
StreamingParser, and since this seems mostly related to the IO layer,
what if we just have special stateful DaffodilInputStream class that
handles this caching of data and other state related to the input. So
something like:
val dis = new DaffodilInputStream(is)
def items : Stream[Node] = {
xmlOut.reset()
val pr = dp.parse(dis, xmlOut)
val item = if (pr.isError) Nil else xmlOut.getResult()
item #:: items
}
So this really is more a change to our I/O layer rather than the
parser/data processors. Another reason why something like this might be
useful is if the data stream was actually something like length-data
pairs, in which case the user might do something like this:
val dis = new DaffodilInputString(is)
def items : Stream[Node] = {
xmlOut.reset()
val len = dis.read() // next byte is a length
val pr = dp.parse(dis, xmlOut, len * 8) // len bytes of data
val item = if (pr.isError) Nil else xmlOut.getResult()
item #:: items
}
Thoughts?
- Steve
