Merge github.com:grpc/grpc into oops-i-split-it-again

8 files changed, 134 insertions, 45 deletions

diff --git a/src/core/json/json.h b/src/core/json/json.h
index 69cbac17dc..b78b42a5b2 100644
--- a/src/core/json/json.h
+++ b/src/core/json/json.h
@@ -60,7 +60,7 @@ typedef struct grpc_json {
  * strings in the tree. The input stream's UTF-8 isn't validated,
  * as in, what you input is what you get as an output.
  *
- * All the keys and values in the grpc_json_t objects will be strings
+ * All the keys and values in the grpc_json objects will be strings
  * pointing at your input buffer.
  *
  * Delete the allocated tree afterward using grpc_json_destroy().
diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c
index 63b4c42131..6156a39d09 100644
--- a/src/core/tsi/ssl_transport_security.c
+++ b/src/core/tsi/ssl_transport_security.c
@@ -54,8 +54,16 @@
 #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_UPPER_BOUND 16384
 #define TSI_SSL_MAX_PROTECTED_FRAME_SIZE_LOWER_BOUND 1024
 
+
+/* Putting a macro like this and littering the source file with #if is really
+   bad practice.
+   TODO(jboeuf): refactor all the #if / #endif in a separate module. */
+#ifndef TSI_OPENSSL_ALPN_SUPPORT
+#define TSI_OPENSSL_ALPN_SUPPORT 1
+#endif
+
 /* TODO(jboeuf): I have not found a way to get this number dynamically from the
- * SSL structure. This is what we would ultimately want though... */
+   SSL structure. This is what we would ultimately want though... */
 #define TSI_SSL_MAX_PROTECTION_OVERHEAD 100
 
 /* --- Structure definitions. ---*/
@@ -70,6 +78,8 @@ struct tsi_ssl_handshaker_factory {
 typedef struct {
   tsi_ssl_handshaker_factory base;
   SSL_CTX* ssl_context;
+  unsigned char* alpn_protocol_list;
+  size_t alpn_protocol_list_length;
 } tsi_ssl_client_handshaker_factory;
 
 typedef struct {
@@ -841,7 +851,7 @@ static tsi_result ssl_handshaker_process_bytes_from_peer(
 static tsi_result ssl_handshaker_extract_peer(tsi_handshaker* self,
                                               tsi_peer* peer) {
   tsi_result result = TSI_OK;
-  const unsigned char* alpn_selected;
+  const unsigned char* alpn_selected = NULL;
   unsigned int alpn_selected_len;
   tsi_ssl_handshaker* impl = (tsi_ssl_handshaker*)self;
   X509* peer_cert = SSL_get_peer_certificate(impl->ssl);
@@ -850,7 +860,14 @@ static tsi_result ssl_handshaker_extract_peer(tsi_handshaker* self,
     X509_free(peer_cert);
     if (result != TSI_OK) return result;
   }
+#if TSI_OPENSSL_ALPN_SUPPORT
   SSL_get0_alpn_selected(impl->ssl, &alpn_selected, &alpn_selected_len);
+#endif /* TSI_OPENSSL_ALPN_SUPPORT */
+  if (alpn_selected == NULL) {
+    /* Try npn. */
+    SSL_get0_next_proto_negotiated(impl->ssl, &alpn_selected,
+                                   &alpn_selected_len);
+  }
   if (alpn_selected != NULL) {
     size_t i;
     tsi_peer_property* new_properties =
@@ -1012,6 +1029,32 @@ static tsi_result create_tsi_ssl_handshaker(SSL_CTX* ctx, int is_client,
   return TSI_OK;
 }
 
+static int select_protocol_list(const unsigned char** out,
+                                unsigned char* outlen,
+                                const unsigned char* client_list,
+                                unsigned int client_list_len,
+                                const unsigned char* server_list,
+                                unsigned int server_list_len) {
+  const unsigned char* client_current = client_list;
+  while ((unsigned int)(client_current - client_list) < client_list_len) {
+    unsigned char client_current_len = *(client_current++);
+    const unsigned char* server_current = server_list;
+    while ((server_current >= server_list) &&
+           (gpr_uintptr)(server_current - server_list) < server_list_len) {
+      unsigned char server_current_len = *(server_current++);
+      if ((client_current_len == server_current_len) &&
+          !memcmp(client_current, server_current, server_current_len)) {
+        *out = server_current;
+        *outlen = server_current_len;
+        return SSL_TLSEXT_ERR_OK;
+      }
+      server_current += server_current_len;
+    }
+    client_current += client_current_len;
+  }
+  return SSL_TLSEXT_ERR_NOACK;
+}
+
 /* --- tsi_ssl__client_handshaker_factory methods implementation. --- */
 
 static tsi_result ssl_client_handshaker_factory_create_handshaker(
@@ -1027,10 +1070,21 @@ static void ssl_client_handshaker_factory_destroy(
     tsi_ssl_handshaker_factory* self) {
   tsi_ssl_client_handshaker_factory* impl =
       (tsi_ssl_client_handshaker_factory*)self;
-  SSL_CTX_free(impl->ssl_context);
+  if (impl->ssl_context != NULL) SSL_CTX_free(impl->ssl_context);
+  if (impl->alpn_protocol_list != NULL) free(impl->alpn_protocol_list);
   free(impl);
 }
 
+static int client_handshaker_factory_npn_callback(
+    SSL* ssl, unsigned char** out, unsigned char* outlen,
+    const unsigned char* in, unsigned int inlen, void* arg) {
+  tsi_ssl_client_handshaker_factory* factory =
+      (tsi_ssl_client_handshaker_factory*)arg;
+  return select_protocol_list((const unsigned char**)out, outlen,
+                              factory->alpn_protocol_list,
+                              factory->alpn_protocol_list_length, in, inlen);
+}
+
 /* --- tsi_ssl_server_handshaker_factory methods implementation. --- */
 
 static tsi_result ssl_server_handshaker_factory_create_handshaker(
@@ -1134,30 +1188,25 @@ static int ssl_server_handshaker_factory_servername_callback(SSL* ssl, int* ap,
   return SSL_TLSEXT_ERR_ALERT_WARNING;
 }
 
+#if TSI_OPENSSL_ALPN_SUPPORT
 static int server_handshaker_factory_alpn_callback(
     SSL* ssl, const unsigned char** out, unsigned char* outlen,
     const unsigned char* in, unsigned int inlen, void* arg) {
   tsi_ssl_server_handshaker_factory* factory =
       (tsi_ssl_server_handshaker_factory*)arg;
-  const unsigned char* client_current = in;
-  while ((unsigned int)(client_current - in) < inlen) {
-    unsigned char client_current_len = *(client_current++);
-    const unsigned char* server_current = factory->alpn_protocol_list;
-    while ((server_current >= factory->alpn_protocol_list) &&
-           (gpr_uintptr)(server_current - factory->alpn_protocol_list) <
-           factory->alpn_protocol_list_length) {
-      unsigned char server_current_len = *(server_current++);
-      if ((client_current_len == server_current_len) &&
-          !memcmp(client_current, server_current, server_current_len)) {
-        *out = server_current;
-        *outlen = server_current_len;
-        return SSL_TLSEXT_ERR_OK;
-      }
-      server_current += server_current_len;
-    }
-    client_current += client_current_len;
-  }
-  return SSL_TLSEXT_ERR_NOACK;
+  return select_protocol_list(out, outlen, in, inlen,
+                              factory->alpn_protocol_list,
+                              factory->alpn_protocol_list_length);
+}
+#endif /* TSI_OPENSSL_ALPN_SUPPORT */
+
+static int server_handshaker_factory_npn_advertised_callback(
+    SSL* ssl, const unsigned char** out, unsigned int* outlen, void* arg) {
+  tsi_ssl_server_handshaker_factory* factory =
+      (tsi_ssl_server_handshaker_factory*)arg;
+  *out = factory->alpn_protocol_list;
+  *outlen = factory->alpn_protocol_list_length;
+  return SSL_TLSEXT_ERR_OK;
 }
 
 /* --- tsi_ssl_handshaker_factory constructors. --- */
@@ -1184,6 +1233,14 @@ tsi_result tsi_create_ssl_client_handshaker_factory(
     gpr_log(GPR_ERROR, "Could not create ssl context.");
     return TSI_INVALID_ARGUMENT;
   }
+
+  impl = calloc(1, sizeof(tsi_ssl_client_handshaker_factory));
+  if (impl == NULL) {
+    SSL_CTX_free(ssl_context);
+    return TSI_OUT_OF_RESOURCES;
+  }
+  impl->ssl_context = ssl_context;
+
   do {
     result =
         populate_ssl_context(ssl_context, pem_private_key, pem_private_key_size,
@@ -1197,41 +1254,33 @@ tsi_result tsi_create_ssl_client_handshaker_factory(
     }
 
     if (num_alpn_protocols != 0) {
-      unsigned char* alpn_protocol_list = NULL;
-      size_t alpn_protocol_list_length = 0;
-      int ssl_failed;
       result = build_alpn_protocol_name_list(
           alpn_protocols, alpn_protocols_lengths, num_alpn_protocols,
-          &alpn_protocol_list, &alpn_protocol_list_length);
+          &impl->alpn_protocol_list, &impl->alpn_protocol_list_length);
       if (result != TSI_OK) {
         gpr_log(GPR_ERROR, "Building alpn list failed with error %s.",
                 tsi_result_to_string(result));
-        free(alpn_protocol_list);
         break;
       }
-      ssl_failed = SSL_CTX_set_alpn_protos(ssl_context, alpn_protocol_list,
-                                           alpn_protocol_list_length);
-      free(alpn_protocol_list);
-      if (ssl_failed) {
+#if TSI_OPENSSL_ALPN_SUPPORT
+      if (SSL_CTX_set_alpn_protos(ssl_context, impl->alpn_protocol_list,
+                                  impl->alpn_protocol_list_length)) {
         gpr_log(GPR_ERROR, "Could not set alpn protocol list to context.");
         result = TSI_INVALID_ARGUMENT;
         break;
       }
+#endif /* TSI_OPENSSL_ALPN_SUPPORT */
+      SSL_CTX_set_next_proto_select_cb(
+          ssl_context, client_handshaker_factory_npn_callback, impl);
     }
   } while (0);
   if (result != TSI_OK) {
-    SSL_CTX_free(ssl_context);
+    ssl_client_handshaker_factory_destroy(&impl->base);
     return result;
   }
   SSL_CTX_set_verify(ssl_context, SSL_VERIFY_PEER, NULL);
   /* TODO(jboeuf): Add revocation verification. */
 
-  impl = calloc(1, sizeof(tsi_ssl_client_handshaker_factory));
-  if (impl == NULL) {
-    SSL_CTX_free(ssl_context);
-    return TSI_OUT_OF_RESOURCES;
-  }
-  impl->ssl_context = ssl_context;
   impl->base.create_handshaker =
       ssl_client_handshaker_factory_create_handshaker;
   impl->base.destroy = ssl_client_handshaker_factory_destroy;
@@ -1322,8 +1371,13 @@ tsi_result tsi_create_ssl_server_handshaker_factory(
           impl->ssl_contexts[i],
           ssl_server_handshaker_factory_servername_callback);
       SSL_CTX_set_tlsext_servername_arg(impl->ssl_contexts[i], impl);
+#if TSI_OPENSSL_ALPN_SUPPORT
       SSL_CTX_set_alpn_select_cb(impl->ssl_contexts[i],
                                  server_handshaker_factory_alpn_callback, impl);
+#endif /* TSI_OPENSSL_ALPN_SUPPORT */
+      SSL_CTX_set_next_protos_advertised_cb(
+          impl->ssl_contexts[i],
+          server_handshaker_factory_npn_advertised_callback, impl);
     } while (0);
 
     if (result != TSI_OK) {
diff --git a/src/ruby/.rubocop_todo.yml b/src/ruby/.rubocop_todo.yml
index c35e970df6..05db404582 100644
--- a/src/ruby/.rubocop_todo.yml
+++ b/src/ruby/.rubocop_todo.yml
@@ -12,7 +12,7 @@ Metrics/AbcSize:
 # Offense count: 3
 # Configuration parameters: CountComments.
 Metrics/ClassLength:
-  Max: 192
+  Max: 200
 
 # Offense count: 35
 # Configuration parameters: CountComments.
diff --git a/src/ruby/bin/interop/interop_client.rb b/src/ruby/bin/interop/interop_client.rb
index 16fb1b199d..da4caa842b 100755
--- a/src/ruby/bin/interop/interop_client.rb
+++ b/src/ruby/bin/interop/interop_client.rb
@@ -284,7 +284,8 @@ class NamedTests
     op = @stub.full_duplex_call(ppp.each_item, return_op: true)
     ppp.canceller_op = op  # causes ppp to cancel after the 1st message
     op.execute.each { |r| ppp.queue.push(r) }
-    assert(op.cancelled, 'call operation should be CANCELLED')
+    op.wait
+    assert(op.cancelled, 'call operation was not CANCELLED')
     p 'OK: cancel_after_first_response'
   end
 
diff --git a/src/ruby/lib/grpc/generic/active_call.rb b/src/ruby/lib/grpc/generic/active_call.rb
index 3814ef34b4..215c0069a3 100644
--- a/src/ruby/lib/grpc/generic/active_call.rb
+++ b/src/ruby/lib/grpc/generic/active_call.rb
@@ -120,6 +120,7 @@ module GRPC
       @started = started
       @unmarshal = unmarshal
       @metadata_tag = metadata_tag
+      @op_notifier = nil
     end
 
     # output_metadata are provides access to hash that can be used to
@@ -148,6 +149,7 @@ module GRPC
     # operation provides a restricted view of this ActiveCall for use as
     # a Operation.
     def operation
+      @op_notifier = Notifier.new
       Operation.new(self)
     end
 
@@ -167,6 +169,7 @@ module GRPC
       batch_result = @call.run_batch(@cq, self, INFINITE_FUTURE, ops)
       return unless assert_finished
       @call.status = batch_result.status
+      op_is_done
       batch_result.check_status
     end
 
@@ -184,6 +187,7 @@ module GRPC
         end
       end
       @call.status = batch_result.status
+      op_is_done
       batch_result.check_status
     end
 
@@ -415,7 +419,7 @@ module GRPC
     def bidi_streamer(requests, **kw, &blk)
       start_call(**kw) unless @started
       bd = BidiCall.new(@call, @cq, @marshal, @unmarshal, @deadline)
-      bd.run_on_client(requests, &blk)
+      bd.run_on_client(requests, @op_notifier, &blk)
     end
 
     # run_server_bidi orchestrates a BiDi stream processing on a server.
@@ -434,6 +438,19 @@ module GRPC
       bd.run_on_server(gen_each_reply)
     end
 
+    # Waits till an operation completes
+    def wait
+      return if @op_notifier.nil?
+      GRPC.logger.debug("active_call.wait: on #{@op_notifier}")
+      @op_notifier.wait
+    end
+
+    # Signals that an operation is done
+    def op_is_done
+      return if @op_notifier.nil?
+      @op_notifier.notify(self)
+    end
+
     private
 
     # Starts the call if not already started
@@ -468,6 +485,6 @@ module GRPC
     # Operation limits access to an ActiveCall's methods for use as
     # a Operation on the client.
     Operation = view_class(:cancel, :cancelled, :deadline, :execute,
-                           :metadata, :status, :start_call)
+                           :metadata, :status, :start_call, :wait)
   end
 end
diff --git a/src/ruby/lib/grpc/generic/bidi_call.rb b/src/ruby/lib/grpc/generic/bidi_call.rb
index 489dd5162a..3b0c71395c 100644
--- a/src/ruby/lib/grpc/generic/bidi_call.rb
+++ b/src/ruby/lib/grpc/generic/bidi_call.rb
@@ -66,6 +66,7 @@ module GRPC
       @cq = q
       @deadline = deadline
       @marshal = marshal
+      @op_notifier = nil  # signals completion on clients
       @readq = Queue.new
       @unmarshal = unmarshal
     end
@@ -76,8 +77,10 @@ module GRPC
     # block that can be invoked with each response.
     #
     # @param requests the Enumerable of requests to send
+    # @op_notifier a Notifier used to signal completion
     # @return an Enumerator of requests to yield
-    def run_on_client(requests, &blk)
+    def run_on_client(requests, op_notifier, &blk)
+      @op_notifier = op_notifier
       @enq_th = Thread.new { write_loop(requests) }
       @loop_th = start_read_loop
       each_queued_msg(&blk)
@@ -105,6 +108,13 @@ module GRPC
     END_OF_READS = :end_of_reads
     END_OF_WRITES = :end_of_writes
 
+    # signals that bidi operation is complete
+    def notify_done
+      return unless @op_notifier
+      GRPC.logger.debug("bidi-notify-done: notifying  #{@op_notifier}")
+      @op_notifier.notify(self)
+    end
+
     # each_queued_msg yields each message on this instances readq
     #
     # - messages are added to the readq by #read_loop
@@ -143,11 +153,13 @@ module GRPC
         @call.status = batch_result.status
         batch_result.check_status
         GRPC.logger.debug("bidi-write-loop: done status #{@call.status}")
+        notify_done
       end
       GRPC.logger.debug('bidi-write-loop: finished')
     rescue StandardError => e
       GRPC.logger.warn('bidi-write-loop: failed')
       GRPC.logger.warn(e)
+      notify_done
       raise e
     end
 
diff --git a/src/ruby/lib/grpc/logconfig.rb b/src/ruby/lib/grpc/logconfig.rb
index 96812170ba..e9b4aa3c95 100644
--- a/src/ruby/lib/grpc/logconfig.rb
+++ b/src/ruby/lib/grpc/logconfig.rb
@@ -38,6 +38,6 @@ Logging.logger.root.appenders = Logging.appenders.stdout
 Logging.logger.root.level = :info
 
 # TODO: provide command-line configuration for logging
-Logging.logger['GRPC'].level = :debug
+Logging.logger['GRPC'].level = :info
 Logging.logger['GRPC::ActiveCall'].level = :info
 Logging.logger['GRPC::BidiCall'].level = :info
diff --git a/test/core/json/json_rewrite_test.c b/test/core/json/json_rewrite_test.c
index ec6deebe76..f5859322ea 100644
--- a/test/core/json/json_rewrite_test.c
+++ b/test/core/json/json_rewrite_test.c
@@ -64,6 +64,11 @@ typedef struct json_reader_userdata {
 static void json_writer_output_char(void* userdata, char c) {
   json_writer_userdata* state = userdata;
   int cmp = fgetc(state->cmp);
+
+  /* treat CRLF as LF */
+  if (cmp == '\r' && c == '\n') {
+    cmp = fgetc(state->cmp);
+  }
   GPR_ASSERT(cmp == c);
 }