GRPC Stream implementation of HeliumStream

blimp/net/grpc_stream.h

Index: blimp/net/grpc_stream.h
diff --git a/blimp/net/grpc_stream.h b/blimp/net/grpc_stream.h
new file mode 100644
index 0000000000000000000000000000000000000000..5158c8cc75f200143f134810bb0d63f5e744a599
--- /dev/null
+++ b/blimp/net/grpc_stream.h
@@ -0,0 +1,108 @@
+// 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 BLIMP_NET_GRPC_STREAM_DELEGATE_H_
+#define BLIMP_NET_GRPC_STREAM_DELEGATE_H_
+
+#include <list>
+
+#include "base/macros.h"
+
+#include "base/callback.h"
+#include "base/command_line.h"
+#include "base/memory/ptr_util.h"
+#include "base/synchronization/lock.h"
+#include "base/threading/thread.h"
+
+#include "blimp/common/proto/helium_service.grpc.pb.h"
+#include "net/base/completion_callback.h"
+
+namespace blimp {
+
+// Callbacks are always invoked in the IO thread.
+class GrpcStream {

[Kevin M, 2016/10/31 21:33:26]

Why doesn't this implement HeliumStream?

+ public:
+  explicit GrpcStream(const net::CompletionCallback& connection_callback);
+
+  // TODO(perumaal): Currently the result is an 'int', but in the future it will
+  // be HeliumResult.
+  using HeliumMessageReceivedCb =
+      base::Callback<void(std::unique_ptr<HeliumWrapper>, int)>;
+
+  using HeliumMessageSentCb = base::Callback<void(int)>;
+
+  // TODO(perumaal): Convert to HeliumStream/HeliumResult next.
+  virtual void SendMessage(std::unique_ptr<HeliumWrapper> helium_message,
+                           const HeliumMessageSentCb& callback) = 0;
+
+  virtual void ReceiveMessage(const HeliumMessageReceivedCb& received_cb) = 0;
+
+  virtual ~GrpcStream();
+
+ protected:
+  class GrpcTag {
+   public:
+    static GrpcTag* Connect(const net::CompletionCallback& connection_cb);
+    static GrpcTag* Write(const GrpcStream::HeliumMessageSentCb& callback);
+    static GrpcTag* Read(const GrpcStream::HeliumMessageReceivedCb& callback);
+    static GrpcTag* WritesDone();
+
+    ~GrpcTag();
+
+    HeliumWrapper* GetReceivedMsg() const;
+
+   private:
+    GrpcTag();
+
+    void ApplyCallbackOnCbThread(bool ok);
+
+    enum class TagType {
+      // Invalid tag type.
+      UNKNOWN = 0,
+      // Tag used during initial connection.
+      CONNECT = 1,
+      // Tag used during waiting for a read to finish.
+      READ = 2,
+      // Tag used during waiting for a single write to finish.
+      WRITE = 3,
+      // Tag used when a sequence of writes has finished.
+      WRITES_DONE = 4
+    };
+
+    TagType tag_type_;
+
+    // For initial connection.
+    net::CompletionCallback connection_cb_;
+
+    // For sending.
+    GrpcStream::HeliumMessageSentCb sent_cb_;
+
+    // For receiving.
+    std::unique_ptr<HeliumWrapper> received_msg_;
+    GrpcStream::HeliumMessageReceivedCb received_cb_;
+
+    friend class GrpcStream;
+  };
+
+  GrpcStream::GrpcTag* GrpcConnectTag();
+  GrpcStream::GrpcTag* GrpcReadTag(const HeliumMessageReceivedCb& received_cb);
+  GrpcStream::GrpcTag* GrpcWriteTag(const HeliumMessageSentCb& sent_cb);
+  GrpcStream::GrpcTag* GrpcWritesDoneTag();
+
+  void StartCompletionQueueThread(grpc::CompletionQueue* completion_q);
+
+ private:
+  static void CompletionQueueThread(
+      grpc::CompletionQueue* completion_q,
+      scoped_refptr<base::SingleThreadTaskRunner> callback_task_runner);
+

[Kevin M, 2016/10/31 21:33:25]

Use a ThreadChecker?

+  scoped_refptr<base::SingleThreadTaskRunner> callback_task_runner_;
+  net::CompletionCallback connection_callback_;
+  std::unique_ptr<base::Thread> grpc_thread_;
+  base::WeakPtrFactory<GrpcStream> weak_factory_;
+};
+
+}  // namespace blimp
+
+#endif  // BLIMP_NET_GRPC_STREAM_DELEGATE_H_