GRPC Stream implementation of HeliumStream

blimp/net/grpc_engine_stream.cc

Index: blimp/net/grpc_engine_stream.cc
diff --git a/blimp/net/grpc_engine_stream.cc b/blimp/net/grpc_engine_stream.cc
new file mode 100644
index 0000000000000000000000000000000000000000..e01ac31a0df07b0c0a05ad878ae7d3438710b840
--- /dev/null
+++ b/blimp/net/grpc_engine_stream.cc
@@ -0,0 +1,54 @@
+// 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_engine_stream.h"
+#include "content/public/browser/browser_thread.h"
+
+#include "base/bind_helpers.h"
+
+namespace blimp {
+
+GrpcEngineStream::GrpcEngineStream(
+    const std::string& ip_address,
+    const net::CompletionCallback& connection_callback)
+    : GrpcStream(connection_callback), ip_address_(ip_address) {
+  grpc::ServerBuilder builder;
+  builder.AddListeningPort(ip_address_, grpc::InsecureServerCredentials());
+  builder.RegisterService(&service_);
+  completion_queue_ = builder.AddCompletionQueue();
+  server_ = builder.BuildAndStart();
+  stream_ = base::MakeUnique<
+      grpc::ServerAsyncReaderWriter<HeliumWrapper, HeliumWrapper>>(&context_);
+  service_.RequestStream(&context_, stream_.get(), completion_queue_.get(),
+                         completion_queue_.get(),
+                         reinterpret_cast<void*>(GrpcConnectTag()));
+  StartCompletionQueueThread(completion_queue_.get());
+  DVLOG(3) << "Starting engine stream @ " << ip_address;
+}
+
+void GrpcEngineStream::SendMessage(
+    std::unique_ptr<HeliumWrapper> helium_message,
+    const Stream::SendMessageCallback& 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 GrpcEngineStream::ReceiveMessage(
+    const Stream::ReceiveMessageCallback& received_cb) {
+  DVLOG(3) << "Receiving (waiting for callback)";
+  GrpcTag* tag = GrpcReadTag(received_cb);
+  stream_->Read(tag->GetReceivedMsg(), reinterpret_cast<void*>(tag));
+}
+
+GrpcEngineStream::~GrpcEngineStream() {
+  LOG(INFO) << "Engine stream shutting down.";
+  server_->Shutdown();
+
+  completion_queue_->Shutdown();
+  completion_queue_.release();
+}
+}  // namespace blimp