[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.
+// Note that any caller must first check the state() of the stream before
+// using it. It is not valid to use a CLOSED stream.
+
+#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 "base/timer/timer.h"
+#include "google/protobuf/io/zero_copy_stream.h"
+#include "google_apis/gcm/base/gcm_export.h"
+
+namespace net {
+class DrainableIOBuffer;
+class IOBufferWithSize;
+class StreamSocket;
+}
+
+namespace gcm {
+
+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,
+    // 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.
+  SocketInputStream(base::TimeDelta read_timeout,
+                    net::StreamSocket* socket);
+  virtual ~SocketInputStream();
+
+  // ZeroCopyInputStream implementation.
+  // Note: the stream must be READY before attempting to read, although it's
+  // valid to read from an EMPTY stream if it's the last read of a message
+  // (see GetNextMessage below).
+  virtual bool Next(const void** data, int* size) OVERRIDE;
+  virtual void BackUp(int count) OVERRIDE;
+  virtual bool Skip(int count) OVERRIDE;
+  virtual int64 ByteCount() const OVERRIDE;

[Ryan Sleevi, 2013/09/10 19:35:27]

STYLE: I thought the policy was NOT to OVERRIDE from third-party libraries (of
which protobuf is). This is same with our Blink no-override policy.

The reasoning being that it makes merges difficult.

Of course, this may be specific to Blink, with its 3 way merges.

[Nicolas Zea, 2013/09/12 22:46:58]

Chromium style guide just says to use OVERRIDE for all implementations (and the
clang style plugins complain otherwise).

+
+  // Reads and appends the next set of data from the socket into the existing
+  // buffer (max read size is dependent on available buffer space).
+  // This is useful to check if there is any pending socket data to be read,
+  // as the read does not time out, so the callback will only be invoked on
+  // an error or when data is received.
+  void Refresh(const base::Closure& callback);
+
+  // Resets the buffer state by copying over any unread data to the beginning
+  // of the buffer and zeroing the buffer position.
+  // Note: it is not valid to call Reset() if state() == CLOSED. The stream
+  // must be recreated in such a scenario.
+  void Reset();

[Ryan Sleevi, 2013/09/10 19:35:27]

STYLE: Naming: I would expect Reset() to actually clear the buffer, but here it
shifts it along, ring-buffer style. Perhaps theres a better name?

[Nicolas Zea, 2013/09/12 22:46:58]

Renamed to Rebuild

+
+  // Asynchronously retrieves the |msg_size| bytes comprising a message,
+  // invoking the callback when complete or upon encountering an error. After a
+  // message is successfully received, Next(..) can be repeatedly invoked to
+  // retrieve the bytes of the message, after which it will return false
+  // (denoting that the message has been fully consumed).
+  // Note: The stream will remain valid after the message is consumed, but
+  // Reset() or another GetNextMessage(..) call must be made before data can be
+  // consumed again.
+  // Also note: GetNextMessage, unlike Refresh, does have a timeout if no
+  // data is received for too long. The timeout will close the stream and
+  // trigger the callback.
+  void GetNextMessage(int msg_size, const base::Closure& callback);
+
+  // Returns the last fatal error encountered. Only valid if state() == CLOSED.
+  int last_error() const;
+
+  // Returns the current state.
+  State state() const;
+
+ private:
+  // Callback for Socket::Read calls.
+  void RefreshCompletionCallback(const base::Closure& callback, int result);
+
+  // Clears the local state.
+  void ResetInternal();
+
+  // Permanently closes the stream.
+  void CloseStream(int error, const base::Closure& callback);
+
+  // Internal net components.
+  net::StreamSocket* socket_;
+  scoped_refptr<net::IOBufferWithSize> io_buffer_;
+  scoped_refptr<net::DrainableIOBuffer> drainable_io_buffer_;
+
+  // The amount of data within |io_buffer_| that has already been read.
+  int buffer_pos_;
+
+  // The amount of valid data within |io_buffer_|.
+  int buffer_size_;

[Ryan Sleevi, 2013/09/10 19:35:27]

STYLE: naming: io_buffer_ is an IOBufferWithSize, so naming this buffer_size_ is
confusing as to which size is the right size.

[Nicolas Zea, 2013/09/12 22:46:58]

Updated to buffer_write_pos_ (and changed buffer_pos_ to buffer_read_pos_ to try
and make that a bit clearer too).

+
+  // The |buffer_pos_| limit at which Next(..) will no longer trigger Refresh
+  // calls.
+  int limit_;
+
+  // The number of bytes we've backed up by. These bytes need to be returned
+  // on the subsequent Next(..) call. 0 <= backup_bytes_ <= buffer_pos_
+  int backup_bytes_;

[Ryan Sleevi, 2013/09/10 19:35:27]

STYLE: Pronouns in comments considered harmful. Or at least confusing. Rephrase
sans "we"

[Nicolas Zea, 2013/09/12 22:46:58]

Done.

+
+  // The number of bytes to skip from the next successful refresh.
+  int skipped_bytes_;
+
+  // If < net::OK, the last net error received.
+  int last_error_;
+
+  // The current state.
+  State state_;
+
+  // Timeout when performing synchronous reads.
+  base::TimeDelta read_timeout_;
+
+  base::OneShotTimer<SocketInputStream> read_timeout_timer_;
+
+  base::WeakPtrFactory<SocketInputStream> weak_ptr_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(SocketInputStream);
+};
+
+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.
+  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.
+  int last_error() const;
+
+ private:
+  void FlushCompletionCallback(const base::Closure& callback, int result);
+
+  net::StreamSocket* socket_;
+  scoped_refptr<net::IOBufferWithSize> io_buffer_;
+  scoped_refptr<net::DrainableIOBuffer> drainable_io_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.
+  int 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_