GRPC Stream implementation of HeliumStream

blimp/net/grpc_client_stream.cc

Index: blimp/net/grpc_client_stream.cc
diff --git a/blimp/net/grpc_client_stream.cc b/blimp/net/grpc_client_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..5a8aef2084615be512195a4a46f47320808bbbb6
--- /dev/null
+++ b/blimp/net/grpc_client_stream.cc
@@ -0,0 +1,53 @@
+// 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.
+
+#include "base/bind_helpers.h"
+#include "blimp/net/grpc_client_stream.h"
+#include "content/public/browser/browser_thread.h"
+
+#include <grpc++/channel.h>
+#include <grpc++/client_context.h>
+#include <grpc++/create_channel.h>
+#include <grpc++/security/credentials.h>
+#include <grpc/grpc.h>
+
+namespace blimp {
+
+GrpcClientStream::GrpcClientStream(
+    const std::string& ip_address,
+    const net::CompletionCallback& connection_callback)
+    : GrpcStream(connection_callback),
+      ip_address_(ip_address),
+      completion_queue_(base::MakeUnique<grpc::CompletionQueue>()) {
+  stub_ = HeliumService::NewStub(
+      grpc::CreateChannel(ip_address, grpc::InsecureChannelCredentials()));
+
+  stream_ =
+      stub_->AsyncStream(&context_, completion_queue_.get(), GrpcConnectTag());
+  StartCompletionQueueThread(completion_queue_.get());
+  DVLOG(3) << "Starting client stream @ " << ip_address;
+}
+
+void GrpcClientStream::SendMessage(
+    std::unique_ptr<HeliumWrapper> helium_message,
+    const HeliumMessageSentCb& sent_cb) {
+  DVLOG(3) << "Sending: " << helium_message->ByteSize() << " bytes";
+  DCHECK(helium_message->serialized_helium_message().size() > 0);
+  stream_->Write(*helium_message.get(),
+                 reinterpret_cast<void*>(GrpcWriteTag(sent_cb)));
+}
+
+void GrpcClientStream::ReceiveMessage(
+    const HeliumMessageReceivedCb& received_cb) {
+  auto tag = GrpcReadTag(received_cb);
+  DVLOG(3) << "Receiving (waiting for callback)";
+  stream_->Read(tag->GetReceivedMsg(), reinterpret_cast<void*>(tag));
+}
+
+GrpcClientStream::~GrpcClientStream() {
+  completion_queue_->Shutdown();
+  completion_queue_.release();
+}
+
+}  // namespace blimp