GRPC Stream implementation of HeliumStream

blimp/net/grpc_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 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 "blimp/helium/stream.h"
+#include "net/base/completion_callback.h"
+
+namespace blimp {
+
+using helium::Stream;
+using helium::Result;
+
+class GrpcStream : public Stream {
+ public:
+  explicit GrpcStream(const net::CompletionCallback& connection_callback);
+
+  ~GrpcStream() override;
+
+ protected:
+  class GrpcTag {
+   public:
+    static GrpcTag* Connect(const net::CompletionCallback& connection_cb);
+    static GrpcTag* Write(const Stream::SendMessageCallback& callback);
+    static GrpcTag* Read(const Stream::ReceiveMessageCallback& 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.
+    Stream::SendMessageCallback sent_cb_;
+
+    // For receiving.
+    std::unique_ptr<HeliumWrapper> received_msg_;
+    Stream::ReceiveMessageCallback received_cb_;
+
+    friend class GrpcStream;
+  };
+
+  GrpcStream::GrpcTag* GrpcConnectTag();
+  GrpcStream::GrpcTag* GrpcReadTag(const Stream::ReceiveMessageCallback& received_cb);
+  GrpcStream::GrpcTag* GrpcWriteTag(const Stream::SendMessageCallback& 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);
+
+  // TODO(perumaal): Use 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_