When one and one makes one

Some time ago I promised to write an article on TGpJoinedStream class. A time has come to fulfill the promise.

TGpJoinedStream is a stream wrapper class – a class that wraps one or more existing streams and provides a TStream interface on the outside. My GpStreams unit contains several examples of such wrappers: TGpStreamWindow provides a stream access to a part of another stream, TGpScatteredStream provides contiguous access to a scattered data (hm, I never wrote an article about that one, didn’t I?), and TGpBufferedStream adds data caching to any stream.

TGpJoinedStream’s role is simple. You pass it a bunch of streams and it provides stream access to concatenated data. IOW, if first stream contains numbers <1, 2, 3> and second stream contains <4, 5>, joined stream would function as if it contains <1, 2, 3, 4, 5>.

Implementation is pretty straightforward and won’t describe it here. I'll rather show how TGpJoinedStream is used in practice and why I wrote it at all.

We have a MPEG parser class that takes a TStream containing MPEG video data and extracts some information from it. It works fine when you need to extract information from a video file, but not so good if you want to use it inside a DirectX filter. From a viewpoint of a DirectX source or sink filter, video is not a stream but a bunch of buffers, which we must process one by one. The trouble is that buffers can almost never be fully processed because MPEG sequences don’t contain data length.

In MPEG, each sequence starts with byte signature 00 00 01, followed by a byte that tells you what kind of data you’re processing, followed by some data. Unless you want to parse each and every sequence data and you know exactly how all well-formed and malformed sequences in the existence are built, you only know that you reached the end of one sequence when you reach another 00 00 01 signature. That’s why the last sequence in the buffer can never be processed (unless this is the last buffer of them all) until we find 00 00 01 in the next buffer. Uff. I hope somebody understands this at all.

In short, we are processing data buffers. A variable-length part of each buffer will stay unprocessed and will have to be prepended to the next buffer before it is passed through the MPEG parser. And so on, until the end of data.

And that’s where TGpJoinedStream comes to help. The unprocessed part is stored in a memory stream FLeftovers, which is empty at the beginning. Buffer parser concatenates FLeftovers with the current data and passes the result through the stream parser. When that one returns, .Position in the combined stream indicates the first unprocessed byte. The unprocessed data is then copied into the FLeftovers stream so it can be used next time the buffer parser is called.

That’s how it looks in code (slightly simplified real code; I just removed some details that are not interesting for our story).

procedure TGpMPEGSequentialParser.ParseNext(buffer: pointer; 
  bufferSize: integer);
var
  combinedStream: TGpJoinedStream;
  mpegParser    : TGpMPEGParser;
  newLeftovers  : TMemoryStream;
  strBuffer     : TGpFixedMemoryStream;
begin
  strBuffer := TGpFixedMemoryStream.Create(buffer^, bufferSize);
  try
    newLeftovers := TMemoryStream.Create;
    try
      combinedStream := TGpJoinedStream.Create([FLeftovers, strBuffer]);
      try
        mpegParser := TGpMPEGParser.Create;
        try
          mpegParser.MPEGStream := combinedStream;

          mpegParser.Parse(HandleMpegSequence);

          if combinedStream.Position < combinedStream.Size then
            newLeftovers.CopyFrom(combinedStream, 
              combinedStream.Size - combinedStream.Position);
        finally FreeAndNil(mpegParser); end;
      finally FreeAndNil(combinedStream); end;
    finally
      FreeAndNil(FLeftovers);
      FLeftovers := newLeftovers;
    end;
  finally FreeAndNil(strBuffer); end;
end;

Hope you like it!