[GCM] Initial work to set up directory structure and introduce socket integration

google_apis/gcm/base/socket_stream.h

+// Copyright (c) 2013 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.
+//
+// Protobuf ZeroCopy[Input/Output]Stream implementations capable of using a
+// net::StreamSocket. Built to work with Protobuf CodedStreams.
+
+#ifndef GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
+#define GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_
+
+#include "base/basictypes.h"
+#include "base/callback_forward.h"
+#include "base/compiler_specific.h"
+#include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
+#include "base/memory/weak_ptr.h"
+#include "google/protobuf/io/zero_copy_stream.h"
+#include "google_apis/gcm/base/gcm_export.h"
+#include "net/base/net_errors.h"
+
+namespace net {
+class DrainableIOBuffer;
+class IOBuffer;
+class StreamSocket;
+}  // namespace net
+
+namespace gcm {
+
+// A helper class for interacting with a net::StreamSocket that is receiving
+// protobuf encoded messages. A SocketInputStream does not take ownership of
+// the socket itself, and it is expected that the life of the input stream
+// should match the life of the socket itself (while the socket remains
+// connected). If an error is encounters, the input stream will store the error
+// in |last_error_|, and state() will be set to CLOSED.
+// Typical usage:
+// 1. Check the state() of the input stream before using it. If CLOSED, the
+//    input stream must be rebuilt (and the socket likely needs to be
+//    reconnected as an error was encountered).
+// 2. If state() is EMPTY, call Refresh(..), passing the maximum byte size for
+//    a message, and wait until completion. It is invalid to attempt to Refresh
+//    an input stream or read data from the stream while a Refresh is pending.
+// 3. Check state() again to ensure the Refresh was successful.
+// 4. Use a CodedInputStream to read from the ZeroCopyInputStream interface of
+//    the SocketInputStream. Next(..) will return true until there is no data
+//    remaining.
+// 5. Call RebuildBuffer when done reading, to shift any unread data to the
+//    start of the buffer.
+// 6. Repeat as necessary.
+class GCM_EXPORT SocketInputStream
+    : public google::protobuf::io::ZeroCopyInputStream {
+ public:
+  enum State {
+    // No valid data to read. This means the buffer is either empty or all data
+    // in the buffer has already been consumed.
+    EMPTY,

[akalin, 2013/10/11 16:47:22]

I'm a fan of having as few states in a state machine as possible. I think you
can eliminate a couple:

- You can remove EMPTY and instead interpret READY && next_pos_ ==
read_buffer_.BytesConsumed() as empty (perhaps add a utility function)
- You can remove CLOSED and instead interpret last_error_ != net::OK as CLOSED.

So typical usage would instead be:

1. Check last_error() before using it...
2. If is_empty() is true, ...
3. Check last_error() again ...
...
What do you think?

[Nicolas Zea, 2013/10/11 18:28:32]

Given that the states are used as a method of exposing internal state rather
than for controlling a state machine, I think I prefer the current approach.
Rather than having to check multiple values in different scenarios, you can
instead check things like state() == READY or state() != READY (which is what
ConnectionHandler does in the larger patch), and then specialize from there.

I think that's more readable/less prone to error.

[akalin, 2013/10/11 18:47:09]

Okay, I see. My main concern is the possibility of the state_ variable going out
of sync with the 'real' state. How about another proposal: keep the State enum,
but make state() figure out the state instead of keeping a separate state_
variable? (and rename it to GetState())

In fact, if we set last_error_ to ERR_IO_PENDING when there's a read/write
pending, then we don't need the state_ variable at all. Then GetState can be:

State GetState() const {
  if (last_error_ < ERR_IO_PENDING)
    return CLOSED;

  if (error == ERR_IO_PENDING)
    return READING;

  DCHECK_EQ(error, OK);
  if (next_pos_ == read_buffer_.BytesConsumed())
    return EMPTY;

  return READY;
}

and the rest of the code can stay the same. What do you think?

[Nicolas Zea, 2013/10/11 23:41:29]

Done.

+    // Valid data to read.
+    READY,
+    // In the process of reading new data from the socket.
+    READING,
+    // An permanent error occurred and the stream is now closed.
+    CLOSED,
+  };
+
+  // |socket| should already be connected.
+  explicit SocketInputStream(net::StreamSocket* socket);
+  virtual ~SocketInputStream();
+
+  // ZeroCopyInputStream implementation.
+  virtual bool Next(const void** data, int* size) OVERRIDE;
+  virtual void BackUp(int count) OVERRIDE;
+  virtual bool Skip(int count) OVERRIDE;  // Not implemented.
+  virtual int64 ByteCount() const OVERRIDE;
+
+  // Reads from the socket, appending a max of |byte_limit| bytes onto the read
+  // buffer. net::ERR_IO_PENDING is returned if the refresh can't complete
+  // synchronously, in which case the callback is invoked upon completion. If
+  // the refresh can complete synchronously returns net::OK (without invoking
+  // callback).
+  // Note: state() (and possibly last_error()) should be checked upon completion
+  // to determine whether the Refresh encountered an error.
+  net::Error Refresh(const base::Closure& callback, int byte_limit);
+
+  // Rebuilds the buffer state by copying over any unread data to the beginning
+  // of the buffer and resetting the buffer read/write positions.
+  // Note: it is not valid to call Rebuild() if state() == CLOSED. The stream
+  // must be recreated from scratch in such a scenario.
+  void RebuildBuffer();
+
+  // Returns the last fatal error encountered. Only valid if state() == CLOSED.
+  net::Error last_error() const;
+
+  // Returns the current state.
+  State state() const;
+
+ private:
+  // Clears the local state.
+  void ResetInternal();
+
+  // Callback for Socket::Read calls.
+  void RefreshCompletionCallback(const base::Closure& callback, int result);
+
+  // Permanently closes the stream.
+  void CloseStream(net::Error error, const base::Closure& callback);
+
+  // Internal net components.
+  net::StreamSocket* const socket_;
+  const scoped_refptr<net::IOBuffer> io_buffer_;
+  // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
+  // been written to yet by Socket::Read calls.
+  const scoped_refptr<net::DrainableIOBuffer> read_buffer_;
+
+  // Starting position of the data within |io_buffer_| to consume on subsequent
+  // Next(..) call.  0 <= next_pos_ <= read_buffer_.BytesConsumed()
+  // Note: next_pos == read_buffer_.BytesConsumed() implies state_ == EMPTY.
+  int next_pos_;
+
+  // If < net::OK, the last net error received.
+  net::Error last_error_;
+
+  // The current state.
+  State state_;
+
+  base::WeakPtrFactory<SocketInputStream> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketInputStream);
+};
+
+// A helper class for writing to a SocketStream with protobuf encoded data.
+// A SocketOutputStream does not take ownership of the socket itself, and it is
+// expected that the life of the output stream should match the life of the
+// socket itself (while the socket remains connected).
+// Typical usage:
+// 1. Check the state() of the output stream before using it. If CLOSED, the
+//    output stream must be rebuilt (and the socket likely needs to be
+//    reconnected, as an error was encountered).
+// 2. If EMPTY, the output stream can be written via a CodedOutputStream using
+//    the ZeroCopyOutputStream interface.
+// 3. Once done writing, state() should be READY, so call Flush(..) to write
+//    the buffer into the StreamSocket. Wait for the callback to be invoked
+//    (it's invalid to write to an output stream while it's flushing).
+// 4. Check the state() again to ensure the Flush was successful. state() should
+//    be EMPTY again.
+// 5. Repeat.
+class GCM_EXPORT SocketOutputStream
+    : public google::protobuf::io::ZeroCopyOutputStream {
+ public:
+  enum State {
+    // No valid data yet.
+    EMPTY,
+    // Ready for flushing (some data is present).
+    READY,
+    // In the process of flushing into the socket.
+    FLUSHING,
+    // A permanent error occurred, and the stream is now closed.
+    CLOSED,
+  };
+
+  // |socket| should already be connected.
+  explicit SocketOutputStream(net::StreamSocket* socket);
+  virtual ~SocketOutputStream();
+
+  // ZeroCopyOutputStream implementation.
+  virtual bool Next(void** data, int* size) OVERRIDE;
+  virtual void BackUp(int count) OVERRIDE;
+  virtual int64 ByteCount() const OVERRIDE;
+
+  // Writes the buffer into the Socket.
+  void Flush(const base::Closure& callback);
+
+  // Returns the current state.
+  State state() const;
+
+  // Returns the last fatal error encountered. Only valid if state() == CLOSED.
+  net::Error last_error() const;
+
+ private:
+  void FlushCompletionCallback(const base::Closure& callback, int result);
+
+  // Internal net components.
+  net::StreamSocket* const socket_;
+  const scoped_refptr<net::IOBuffer> io_buffer_;
+  // IOBuffer implementation that wraps the data within |io_buffer_| that hasn't
+  // been written to the socket yet.
+  const scoped_refptr<net::DrainableIOBuffer> write_buffer_;
+
+  // The amount of valid buffer data. This is the amount of data written on the
+  // next call to Flush().
+  int buffer_used_;
+
+  // If < net::OK, the last net error received.
+  net::Error last_error_;
+
+  // The current state.
+  State state_;
+
+  base::WeakPtrFactory<SocketOutputStream> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketOutputStream);
+};
+
+}  // namespace gcm
+
+#endif  // GOOGLE_APIS_GCM_BASE_SOCKET_STREAM_H_