Add ServiceAccount Credentials wrapping and handle credentials creation

failure.
	Change on 2015/01/09 by yangg <yangg@google.com>
-------------
Created by MOE: http://code.google.com/p/moe-java
MOE_MIGRATED_REVID=83634736

5 files changed, 81 insertions, 11 deletions

diff --git a/src/core/security/security_context.c b/src/core/security/security_context.c
index 13b9a847ee..917a22f453 100644
--- a/src/core/security/security_context.c
+++ b/src/core/security/security_context.c
@@ -451,7 +451,7 @@ static grpc_channel *grpc_ssl_channel_create(grpc_credentials *creds,
   grpc_security_status status = GRPC_SECURITY_OK;
   size_t i = 0;
   const char *secure_peer_name = target;
-  for (i = 0; i < args->num_args; i++) {
+  for (i = 0; args && i < args->num_args; i++) {
     grpc_arg *arg = &args->args[i];
     if (!strcmp(arg->key, GRPC_SSL_TARGET_NAME_OVERRIDE_ARG) &&
         arg->type == GRPC_ARG_STRING) {
@@ -492,12 +492,17 @@ static grpc_channel *grpc_channel_create_from_composite_creds(
     return grpc_ssl_channel_create(
         composite_creds, grpc_ssl_credentials_get_config(creds), target, args);
   }
-  return NULL; /* TODO(ctiller): return lame channel. */
+  gpr_log(GPR_ERROR, "Credentials is insufficient to create a secure channel.");
+  return grpc_lame_client_channel_create();
 }
 
 grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
                                          const char *target,
                                          const grpc_channel_args *args) {
+  if (creds == NULL) {
+    gpr_log(GPR_ERROR, "No credentials to create a secure channel.");
+    return grpc_lame_client_channel_create();
+  }
   if (grpc_credentials_has_request_metadata_only(creds)) {
     gpr_log(GPR_ERROR,
             "Credentials is insufficient to create a secure channel.");
@@ -518,7 +523,8 @@ grpc_channel *grpc_secure_channel_create(grpc_credentials *creds,
     return grpc_channel_create_from_composite_creds(creds, target, args);
   } else {
     gpr_log(GPR_ERROR,
-            "Unknown credentials type %s for creating a secure channel.");
+            "Unknown credentials type %s for creating a secure channel.",
+            creds->type);
     return grpc_lame_client_channel_create();
   }
 }
diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c
index 6a832436ca..5fa3e42362 100644
--- a/src/core/surface/lame_client.c
+++ b/src/core/surface/lame_client.c
@@ -33,6 +33,8 @@
 
 #include "src/core/surface/lame_client.h"
 
+#include <string.h>
+
 #include "src/core/channel/channel_stack.h"
 #include "src/core/surface/channel.h"
 #include "src/core/surface/call.h"
@@ -42,16 +44,28 @@
 
 typedef struct { void *unused; } call_data;
 
-typedef struct { void *unused; } channel_data;
+typedef struct { grpc_mdelem *message; } channel_data;
+
+static void do_nothing(void *data, grpc_op_error error) {}
 
 static void call_op(grpc_call_element *elem, grpc_call_element *from_elem,
                     grpc_call_op *op) {
+  channel_data *channeld = elem->channel_data;
   GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
 
   switch (op->type) {
-    case GRPC_SEND_START:
+    case GRPC_SEND_START: {
+      grpc_call_op set_status_op;
+      grpc_mdelem_ref(channeld->message);
+      memset(&set_status_op, 0, sizeof(grpc_call_op));
+      set_status_op.dir = GRPC_CALL_UP;
+      set_status_op.type = GRPC_RECV_METADATA;
+      set_status_op.done_cb = do_nothing;
+      set_status_op.data.metadata = channeld->message;
+      grpc_call_recv_metadata(elem, &set_status_op);
       grpc_call_recv_finish(elem, 1);
       break;
+    }
     case GRPC_SEND_METADATA:
       grpc_mdelem_unref(op->data.metadata);
       break;
@@ -81,11 +95,20 @@ static void destroy_call_elem(grpc_call_element *elem) {}
 static void init_channel_elem(grpc_channel_element *elem,
                               const grpc_channel_args *args, grpc_mdctx *mdctx,
                               int is_first, int is_last) {
+  channel_data *channeld = elem->channel_data;
+
   GPR_ASSERT(is_first);
   GPR_ASSERT(is_last);
+
+  channeld->message = grpc_mdelem_from_strings(mdctx, "grpc-message",
+                                               "Rpc sent on a lame channel.");
 }
 
-static void destroy_channel_elem(grpc_channel_element *elem) {}
+static void destroy_channel_elem(grpc_channel_element *elem) {
+  channel_data *channeld = elem->channel_data;
+
+  grpc_mdelem_unref(channeld->message);
+}
 
 static const grpc_channel_filter lame_filter = {
     call_op, channel_op,
diff --git a/src/cpp/client/channel.cc b/src/cpp/client/channel.cc
index 7d95518631..6c8879d577 100644
--- a/src/cpp/client/channel.cc
+++ b/src/cpp/client/channel.cc
@@ -69,8 +69,9 @@ Channel::Channel(const grpc::string& target,
                   : args.GetSslTargetNameOverride()) {
   grpc_channel_args channel_args;
   args.SetChannelArgs(&channel_args);
+  grpc_credentials* c_creds = creds ? creds->GetRawCreds() : nullptr;
   c_channel_ = grpc_secure_channel_create(
-      creds->GetRawCreds(), target.c_str(),
+      c_creds, target.c_str(),
       channel_args.num_args > 0 ? &channel_args : nullptr);
 }
 
@@ -118,10 +119,15 @@ Status Channel::StartBlockingRpc(const RpcMethod& method,
                                     finished_tag,
                                     GRPC_WRITE_BUFFER_HINT) == GRPC_CALL_OK);
   ev = grpc_completion_queue_pluck(cq, invoke_tag, gpr_inf_future);
+  bool success = ev->data.invoke_accepted == GRPC_OP_OK;
   grpc_event_finish(ev);
+  if (!success) {
+    GetFinalStatus(cq, finished_tag, &status);
+    return status;
+  }
   // write request
   grpc_byte_buffer* write_buffer = nullptr;
-  bool success = SerializeProto(request, &write_buffer);
+  success = SerializeProto(request, &write_buffer);
   if (!success) {
     grpc_call_cancel(call);
     status =
diff --git a/src/cpp/client/credentials.cc b/src/cpp/client/credentials.cc
index 986008f7bb..cac1d3d106 100644
--- a/src/cpp/client/credentials.cc
+++ b/src/cpp/client/credentials.cc
@@ -35,6 +35,7 @@
 #include <string>
 
 #include <grpc/grpc_security.h>
+#include <grpc/support/log.h>
 
 #include <grpc++/credentials.h>
 
@@ -58,6 +59,9 @@ std::unique_ptr<Credentials> CredentialsFactory::SslCredentials(
       options.pem_root_certs.empty() ? nullptr
                                      : reinterpret_cast<const unsigned char*>(
                                            options.pem_root_certs.c_str());
+  if (pem_root_certs == nullptr) {
+    return std::unique_ptr<Credentials>();
+  }
   const unsigned char* pem_private_key =
       options.pem_private_key.empty() ? nullptr
                                       : reinterpret_cast<const unsigned char*>(
@@ -71,14 +75,40 @@ std::unique_ptr<Credentials> CredentialsFactory::SslCredentials(
       pem_root_certs, options.pem_root_certs.size(), pem_private_key,
       options.pem_private_key.size(), pem_cert_chain,
       options.pem_cert_chain.size());
-  std::unique_ptr<Credentials> cpp_creds(new Credentials(c_creds));
+  std::unique_ptr<Credentials> cpp_creds(
+      c_creds == nullptr ? nullptr : new Credentials(c_creds));
   return cpp_creds;
 }
 
 // Builds credentials for use when running in GCE
 std::unique_ptr<Credentials> CredentialsFactory::ComputeEngineCredentials() {
   grpc_credentials* c_creds = grpc_compute_engine_credentials_create();
-  std::unique_ptr<Credentials> cpp_creds(new Credentials(c_creds));
+  std::unique_ptr<Credentials> cpp_creds(
+      c_creds == nullptr ? nullptr : new Credentials(c_creds));
+  return cpp_creds;
+}
+
+// Builds service account credentials.
+std::unique_ptr<Credentials> CredentialsFactory::ServiceAccountCredentials(
+    const grpc::string& json_key, const grpc::string& scope,
+    std::chrono::seconds token_lifetime) {
+  gpr_timespec lifetime = gpr_time_from_seconds(
+      token_lifetime.count() > 0 ? token_lifetime.count() : 0);
+  grpc_credentials* c_creds = grpc_service_account_credentials_create(
+      json_key.c_str(), scope.c_str(), lifetime);
+  std::unique_ptr<Credentials> cpp_creds(
+      c_creds == nullptr ? nullptr : new Credentials(c_creds));
+  return cpp_creds;
+}
+
+// Builds IAM credentials.
+std::unique_ptr<Credentials> CredentialsFactory::IAMCredentials(
+    const grpc::string& authorization_token,
+    const grpc::string& authority_selector) {
+  grpc_credentials* c_creds = grpc_iam_credentials_create(
+      authorization_token.c_str(), authority_selector.c_str());
+  std::unique_ptr<Credentials> cpp_creds(
+      c_creds == nullptr ? nullptr : new Credentials(c_creds));
   return cpp_creds;
 }
 
@@ -93,7 +123,8 @@ std::unique_ptr<Credentials> CredentialsFactory::ComposeCredentials(
   // refcounts incremented.
   grpc_credentials* c_creds = grpc_composite_credentials_create(
       creds1->GetRawCreds(), creds2->GetRawCreds());
-  std::unique_ptr<Credentials> cpp_creds(new Credentials(c_creds));
+  std::unique_ptr<Credentials> cpp_creds(
+      c_creds == nullptr ? nullptr : new Credentials(c_creds));
   return cpp_creds;
 }
 
diff --git a/src/cpp/stream/stream_context.cc b/src/cpp/stream/stream_context.cc
index 22b7e7d494..b55e647a63 100644
--- a/src/cpp/stream/stream_context.cc
+++ b/src/cpp/stream/stream_context.cc
@@ -85,6 +85,10 @@ void StreamContext::Start(bool buffered) {
     GPR_ASSERT(GRPC_CALL_OK == error);
     grpc_event* invoke_ev =
         grpc_completion_queue_pluck(cq(), invoke_tag(), gpr_inf_future);
+    if (invoke_ev->data.invoke_accepted != GRPC_OP_OK) {
+      peer_halfclosed_ = true;
+      self_halfclosed_ = true;
+    }
     grpc_event_finish(invoke_ev);
   } else {
     // TODO(yangg) metadata needs to be added before accept