Add net::SourceStream and net::FilterSourceStream

net/filter/filter_source_stream.h

+// Copyright 2016 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef NET_FILTER_FILTER_SOURCE_STREAM_H_
+#define NET_FILTER_FILTER_SOURCE_STREAM_H_
+
+#include <memory>
+#include <string>
+
+#include "base/macros.h"
+#include "base/memory/ref_counted.h"
+#include "net/base/completion_callback.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_export.h"
+#include "net/filter/source_stream.h"
+
+namespace net {
+
+class DrainableIOBuffer;
+class IOBuffer;
+
+// FilterSourceStream represents SourceStreams that always have an upstream
+// from which undecoded input can be read. Except the ultimate upstream in
+// the filter chain, all other streams should implement FilterSourceStream
+// instead of SourceStream.
+class NET_EXPORT_PRIVATE FilterSourceStream : public SourceStream {
+ public:
+  // |upstream| is the SourceStream from which |this| will read data.
+  // |upstream| cannot be null.
+  FilterSourceStream(SourceType type, std::unique_ptr<SourceStream> upstream);
+
+  ~FilterSourceStream() override;
+
+  int Read(IOBuffer* read_buffer,
+           int read_buffer_size,
+           const CompletionCallback& callback) override;
+
+  std::string Description() const override;
+
+ private:
+  enum State {
+    STATE_NONE,
+    // Reading data from |upstream_| into |input_buffer_|.
+    STATE_READ_DATA,
+    // Reading data from |upstream_| completed.
+    STATE_READ_DATA_COMPLETE,
+    // Filtering data contained in |input_buffer_|.
+    STATE_FILTER_DATA,
+    // Filtering data contained in |input_buffer_| completed.
+    STATE_FILTER_DATA_COMPLETE,
+    STATE_DONE,
+  };
+
+  int DoLoop(int result);
+  int DoReadData();
+  int DoReadDataComplete(int result);
+  int DoFilterData();
+
+  // Helper method used as a callback argument passed to |upstream_->Read()|.
+  void OnIOComplete(int result);
+
+  // Subclasses should implement this method to filter data from
+  // |input_buffer| and write to |output_buffer|.
+  // This method must complete synchronously (i.e. It cannot return
+  // ERR_IO_PENDING). If an unrecoverable error occurred, this should return
+  // ERR_CONTENT_DECODING_FAILED or a more specific error code.
+  //
+  // FilterData() will be repeatedly invoked with the same |input_buffer| until

[Randy Smith (Not in Mondays), 2016/09/12 21:05:03]

Could we remove the phrase "with the same |input_buffer|"?  That seems like
something that we don't want the subclass to be counting on.  All we want to
guarantee is that we (superclass) won't lose any data they (subclass) haven't
drained.

(Suggestion: Matt raised the possibility of an interface where instead of
passing a DrainableIOBuffer, we pass an IOBuffer & size, so it's a pure input
argument rather than an input/output argument.  This would also have the benefit
of not allowing the possibility of the subclass assuming that we're passing the
same IOBuffer, since we wouldn't be--we'd rewrap on each call.  I like this
idea, but not with a deep passion, so just a suggestion.)

[xunjieli, 2016/09/13 19:45:24]

I added this comment per an earlier review commment. I agree this is an
implementation detail, which probably shouldn't be in the method documentation.
However, I feel that this is an important detail of the limitation of the
current design. I will keep it unless you feel strongly about removing it.
Acknowledged. I will keep this as it is. I like the DrainableIOBuffer so
subclasses can have multiple routines to consume bytes from the buffer without
keeping track of the offsets themselves. You can see gzip_source_stream.cc for
an example. I will keep this as it is.

[mmenke, 2016/09/13 19:50:57]

Don't think you understand the comment - the parent class would track the
offset, we'd just remove the weird contract that we always return the same
drainable IO buffer, which the subclass is responsible for managing itself.

i.e, we'd do something like:

FilterData(..., in_buffer, in_buffer_size, &consumed_bytes);
which then reads part of the data and sets consumed_bytes to if (say it's set to
8).

Then the next call to FilterData would return another IOBuffer, with 8 fewer
bytes in it (i.e., the parent class would completely manage the
DrainableIOBuffer).

[xunjieli, 2016/09/13 19:54:58]

Thanks for the follow-up. I will do that in a separate CL.

+  // FilterData() returns 0 or an error. If FilterData() returns 0,
+  // |input_buffer| must be fully drained. Upstream EOF is reached when
+  // FilterData() is called with |input_buffer->BytesRemaining() == 0|.

[Randy Smith (Not in Mondays), 2016/09/12 21:05:03]

Helen, as I mentioned offline I think there's a problem with this interface, as
I don't think we can guarantee that FilterData() will never be called with
|input_buffer->BytesRemaining() == 0| in the non-EOF case.  If FilterData drains
everything from input_buffer, fills up output_buffer, and has stuff left over,
we need to provide some way for it to continue to output the data, which I think
means calling it again with input_buffer->BytesRemaining() == 0.  

Would it be reasonable to go back to the earlier suggestion of having a separate
interface for signaling EOF?  I'm inclined to make this dirt simple:

virtual int UpstreamEof() { return OK; }

And include in the interface contract that it can't be called until FilterData()
has returned 0 (indicating that it doesn't have any more data to return without
further input from the upstream).  In this case I think it's reasonable for
many/most subclasses not to bother to implement UpstreamEof(); they decompress
on the fly, and whenever the upstream is done, they are also done.  And if they
have more bufferred data to output, they'll output it on the previous calls to
FilterData().  WDYT?

[xunjieli, 2016/09/13 19:45:24]

Done. I will upload this change as a separate CL.

+  // TODO(xunjieli): consider allowing asynchronous response via callback
+  // to support off-thread decompression.
+  virtual int FilterData(IOBuffer* output_buffer,
+                         int output_buffer_size,
+                         DrainableIOBuffer* input_buffer) = 0;
+
+  // Returns a string representation of the type of this FilterSourceStream.
+  // This is for UMA logging.
+  virtual std::string GetTypeAsString() const = 0;
+
+  // Returns whether |this| still needs more input data from |upstream_|.
+  // By default, |this| will continue reading until |upstream_| returns an error
+  // or EOF. Subclass can override this to return false to skip reading all the
+  // input from |upstream_|.
+  virtual bool NeedMoreData() const;
+
+  // The SourceStream from which |this| will read data from. Data flows from
+  // |upstream_| to |this_|.
+  std::unique_ptr<SourceStream> upstream_;
+
+  State next_state_;
+
+  // Buffer for reading data out of |upstream_| and then for use by |this|
+  // before the filtered data is returned through Read().
+  scoped_refptr<IOBuffer> input_buffer_;
+
+  // Wrapper around |input_buffer_| that makes visible only the unread data.
+  // Keep this as a member because subclass might not drain everything in a
+  // single FilterData().
+  scoped_refptr<DrainableIOBuffer> drainable_input_buffer_;
+
+  // Not null if there is a pending Read.
+  scoped_refptr<IOBuffer> output_buffer_;
+  int output_buffer_size_;
+  CompletionCallback callback_;
+
+  // Reading from |upstream_| has returned 0 byte or an error code.
+  bool upstream_end_reached_;
+
+  DISALLOW_COPY_AND_ASSIGN(FilterSourceStream);
+};
+
+}  // namespace net
+
+#endif  // NET_FILTER_FILTER_SOURCE_STREAM_H_