google_apis/gcm/engine/connection_handler_impl.h

   1 // Copyright 2013 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #ifndef GOOGLE_APIS_GCM_ENGINE_CONNECTION_HANDLER_IMPL_H_
   6 #define GOOGLE_APIS_GCM_ENGINE_CONNECTION_HANDLER_IMPL_H_
   7
   8 #include <vector>
   9
  10 #include "base/memory/weak_ptr.h"
  11 #include "base/time/time.h"
  12 #include "base/timer/timer.h"
  13 #include "google_apis/gcm/engine/connection_handler.h"
  14
  15 namespace mcs_proto {
  16 class LoginRequest;
  17 }  // namespace mcs_proto
  18
  19 namespace gcm {
  20
  21 class GCM_EXPORT ConnectionHandlerImpl : public ConnectionHandler {
  22  public:
  23   // |read_callback| will be invoked with the contents of any received protobuf
  24   // message.
  25   // |write_callback| will be invoked anytime a message has been successfully
  26   // sent. Note: this just means the data was sent to the wire, not that the
  27   // other end received it.
  28   // |connection_callback| will be invoked with any fatal read/write errors
  29   // encountered.
  30   ConnectionHandlerImpl(
  31       base::TimeDelta read_timeout,
  32       const ProtoReceivedCallback& read_callback,
  33       const ProtoSentCallback& write_callback,
  34       const ConnectionChangedCallback& connection_callback);
  35   ~ConnectionHandlerImpl() override;
  36
  37   // ConnectionHandler implementation.
  38   void Init(const mcs_proto::LoginRequest& login_request,
  39             net::StreamSocket* socket) override;
  40   void Reset() override;
  41   bool CanSendMessage() const override;
  42   void SendMessage(const google::protobuf::MessageLite& message) override;
  43
  44  private:
  45   // State machine for handling incoming data. See WaitForData(..) for usage.
  46   enum ProcessingState {
  47     // Processing the version, tag, and size packets (assuming minimum length
  48     // size packet). Only used during the login handshake.
  49     MCS_VERSION_TAG_AND_SIZE = 0,
  50     // Processing the tag and size packets (assuming minimum length size
  51     // packet). Used for normal messages.
  52     MCS_TAG_AND_SIZE,
  53     // Processing the size packet alone.
  54     MCS_SIZE,
  55     // Processing the protocol buffer bytes (for those messages with non-zero
  56     // sizes).
  57     MCS_PROTO_BYTES
  58   };
  59
  60   // Sends the protocol version and login request. First step in the MCS
  61   // connection handshake.
  62   void Login(const google::protobuf::MessageLite& login_request);
  63
  64   // SendMessage continuation. Invoked when Socket::Write completes.
  65   void OnMessageSent();
  66
  67   // Starts the message processing process, which is comprised of the tag,
  68   // message size, and bytes packet types.
  69   void GetNextMessage();
  70
  71   // Performs any necessary SocketInputStream refreshing until the data
  72   // associated with |packet_type| is fully ready, then calls the appropriate
  73   // OnGot* message to process the packet data. If the read times out,
  74   // will close the stream and invoke the connection callback.
  75   void WaitForData(ProcessingState state);
  76
  77   // Incoming data helper methods.
  78   void OnGotVersion();
  79   void OnGotMessageTag();
  80   void OnGotMessageSize();
  81   void OnGotMessageBytes();
  82
  83   // Timeout handler.
  84   void OnTimeout();
  85
  86   // Closes the current connection.
  87   void CloseConnection();
  88
  89   // Timeout policy: the timeout is only enforced while waiting on the
  90   // handshake (version and/or LoginResponse) or once at least a tag packet has
  91   // been received. It is reset every time new data is received, and is
  92   // only stopped when a full message is processed.
  93   // TODO(zea): consider enforcing a separate timeout when waiting for
  94   // a message to send.
  95   const base::TimeDelta read_timeout_;
  96   base::OneShotTimer<ConnectionHandlerImpl> read_timeout_timer_;
  97
  98   // This connection's socket and the input/output streams attached to it.
  99   net::StreamSocket* socket_;
 100   scoped_ptr<SocketInputStream> input_stream_;
 101   scoped_ptr<SocketOutputStream> output_stream_;
 102
 103   // Whether the MCS login handshake has successfully completed. See Init(..)
 104   // description for more info on what the handshake involves.
 105   bool handshake_complete_;
 106
 107   // State for the message currently being processed, if there is one.
 108   uint8 message_tag_;
 109   uint32 message_size_;
 110
 111   ProtoReceivedCallback read_callback_;
 112   ProtoSentCallback write_callback_;
 113   ConnectionChangedCallback connection_callback_;
 114
 115   // The number of bytes of the size packet read so far without finishing the
 116   // read. This can be up to but no larger than 5 (the max number of bytes for
 117   // a varint32).
 118   uint8 size_packet_so_far_;
 119   // A temporary input buffer for holding large messages. Will hold
 120   // message_size_ bytes for messages larger than the normal size limit (and
 121   // will be empty otherwise).
 122   std::vector<uint8> payload_input_buffer_;
 123
 124   base::WeakPtrFactory<ConnectionHandlerImpl> weak_ptr_factory_;
 125
 126   DISALLOW_COPY_AND_ASSIGN(ConnectionHandlerImpl);
 127 };
 128
 129 }  // namespace gcm
 130
 131 #endif  // GOOGLE_APIS_GCM_ENGINE_CONNECTION_HANDLER_IMPL_H_