test/core/end2end/tests/stream_compression_compressed_payload.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include "test/core/end2end/end2end_tests.h"
  20
  21 #include <stdio.h>
  22 #include <string.h>
  23
  24 #include <string>
  25
  26 #include "absl/strings/str_cat.h"
  27
  28 #include <grpc/byte_buffer.h>
  29 #include <grpc/byte_buffer_reader.h>
  30 #include <grpc/compression.h>
  31 #include <grpc/support/alloc.h>
  32 #include <grpc/support/log.h>
  33 #include <grpc/support/time.h>
  34
  35 #include "src/core/lib/channel/channel_args.h"
  36 #include "src/core/lib/compression/compression_args.h"
  37 #include "src/core/lib/surface/call.h"
  38 #include "src/core/lib/surface/call_test_only.h"
  39 #include "src/core/lib/transport/static_metadata.h"
  40 #include "test/core/end2end/cq_verifier.h"
  41
  42 static void* tag(intptr_t t) { return reinterpret_cast<void*>(t); }
  43
  44 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
  45                                             const char* test_name,
  46                                             grpc_channel_args* client_args,
  47                                             grpc_channel_args* server_args) {
  48   grpc_end2end_test_fixture f;
  49   gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
  50   f = config.create_fixture(client_args, server_args);
  51   config.init_server(&f, server_args);
  52   config.init_client(&f, client_args);
  53   return f;
  54 }
  55
  56 static gpr_timespec n_seconds_from_now(int n) {
  57   return grpc_timeout_seconds_to_deadline(n);
  58 }
  59
  60 static gpr_timespec five_seconds_from_now(void) {
  61   return n_seconds_from_now(5);
  62 }
  63
  64 static void drain_cq(grpc_completion_queue* cq) {
  65   grpc_event ev;
  66   do {
  67     ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
  68   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
  69 }
  70
  71 static void shutdown_server(grpc_end2end_test_fixture* f) {
  72   if (!f->server) return;
  73   grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
  74   GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
  75                                          grpc_timeout_seconds_to_deadline(5),
  76                                          nullptr)
  77                  .type == GRPC_OP_COMPLETE);
  78   grpc_server_destroy(f->server);
  79   f->server = nullptr;
  80 }
  81
  82 static void shutdown_client(grpc_end2end_test_fixture* f) {
  83   if (!f->client) return;
  84   grpc_channel_destroy(f->client);
  85   f->client = nullptr;
  86 }
  87
  88 static void end_test(grpc_end2end_test_fixture* f) {
  89   shutdown_server(f);
  90   shutdown_client(f);
  91
  92   grpc_completion_queue_shutdown(f->cq);
  93   drain_cq(f->cq);
  94   grpc_completion_queue_destroy(f->cq);
  95   grpc_completion_queue_destroy(f->shutdown_cq);
  96 }
  97
  98 static void request_for_disabled_algorithm(
  99     grpc_end2end_test_config config, const char* test_name,
 100     uint32_t send_flags_bitmask,
 101     grpc_compression_algorithm algorithm_to_disable,
 102     grpc_compression_algorithm requested_client_compression_algorithm,
 103     grpc_status_code expected_error, grpc_metadata* client_metadata) {
 104   grpc_call* c;
 105   grpc_call* s;
 106   grpc_slice request_payload_slice;
 107   grpc_byte_buffer* request_payload;
 108   grpc_channel_args* client_args;
 109   grpc_channel_args* server_args;
 110   grpc_end2end_test_fixture f;
 111   grpc_op ops[6];
 112   grpc_op* op;
 113   grpc_metadata_array initial_metadata_recv;
 114   grpc_metadata_array trailing_metadata_recv;
 115   grpc_metadata_array request_metadata_recv;
 116   grpc_byte_buffer* request_payload_recv = nullptr;
 117   grpc_call_details call_details;
 118   grpc_status_code status;
 119   grpc_call_error error;
 120   grpc_slice details;
 121   int was_cancelled = 2;
 122   cq_verifier* cqv;
 123   char str[1024];
 124
 125   memset(str, 'x', 1023);
 126   str[1023] = '\0';
 127   request_payload_slice = grpc_slice_from_copied_string(str);
 128   request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 129
 130   client_args = grpc_channel_args_set_channel_default_compression_algorithm(
 131       nullptr, requested_client_compression_algorithm);
 132   server_args = grpc_channel_args_set_channel_default_compression_algorithm(
 133       nullptr, GRPC_COMPRESS_NONE);
 134   {
 135     grpc_core::ExecCtx exec_ctx;
 136     server_args = grpc_channel_args_compression_algorithm_set_state(
 137         &server_args, algorithm_to_disable, false);
 138   }
 139
 140   f = begin_test(config, test_name, client_args, server_args);
 141   cqv = cq_verifier_create(f.cq);
 142
 143   gpr_timespec deadline = five_seconds_from_now();
 144   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 145                                grpc_slice_from_static_string("/foo"), nullptr,
 146                                deadline, nullptr);
 147   GPR_ASSERT(c);
 148
 149   grpc_metadata_array_init(&initial_metadata_recv);
 150   grpc_metadata_array_init(&trailing_metadata_recv);
 151   grpc_metadata_array_init(&request_metadata_recv);
 152   grpc_call_details_init(&call_details);
 153
 154   error =
 155       grpc_server_request_call(f.server, &s, &call_details,
 156                                &request_metadata_recv, f.cq, f.cq, tag(101));
 157   GPR_ASSERT(GRPC_CALL_OK == error);
 158
 159   memset(ops, 0, sizeof(ops));
 160   op = ops;
 161   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 162   if (client_metadata != nullptr) {
 163     op->data.send_initial_metadata.count = 1;
 164     op->data.send_initial_metadata.metadata = client_metadata;
 165   } else {
 166     op->data.send_initial_metadata.count = 0;
 167   }
 168   op->flags = 0;
 169   op->reserved = nullptr;
 170   op++;
 171   op->op = GRPC_OP_SEND_MESSAGE;
 172   op->data.send_message.send_message = request_payload;
 173   op->flags = send_flags_bitmask;
 174   op->reserved = nullptr;
 175   op++;
 176   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 177   op->flags = 0;
 178   op->reserved = nullptr;
 179   op++;
 180   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 181   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 182   op->flags = 0;
 183   op->reserved = nullptr;
 184   op++;
 185   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 186   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 187   op->data.recv_status_on_client.status = &status;
 188   op->data.recv_status_on_client.status_details = &details;
 189   op->flags = 0;
 190   op->reserved = nullptr;
 191   op++;
 192   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 193                                 nullptr);
 194   GPR_ASSERT(GRPC_CALL_OK == error);
 195
 196   CQ_EXPECT_COMPLETION(cqv, tag(101), true);
 197   CQ_EXPECT_COMPLETION(cqv, tag(1), true);
 198   cq_verify(cqv);
 199
 200   op = ops;
 201   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 202   op->data.send_initial_metadata.count = 0;
 203   op->flags = 0;
 204   op->reserved = nullptr;
 205   op++;
 206   op->op = GRPC_OP_RECV_MESSAGE;
 207   op->data.recv_message.recv_message = &request_payload_recv;
 208   op->flags = 0;
 209   op->reserved = nullptr;
 210   op++;
 211   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(102),
 212                                 nullptr);
 213   GPR_ASSERT(GRPC_CALL_OK == error);
 214
 215   CQ_EXPECT_COMPLETION(cqv, tag(102), false);
 216
 217   op = ops;
 218   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 219   op->data.recv_close_on_server.cancelled = &was_cancelled;
 220   op->flags = 0;
 221   op->reserved = nullptr;
 222   op++;
 223   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(103),
 224                                 nullptr);
 225   GPR_ASSERT(GRPC_CALL_OK == error);
 226
 227   CQ_EXPECT_COMPLETION(cqv, tag(103), true);
 228   cq_verify(cqv);
 229
 230   /* call was cancelled (closed) ... */
 231   GPR_ASSERT(was_cancelled != 0);
 232   /* with a certain error */
 233   GPR_ASSERT(status == expected_error);
 234
 235   const char* algo_name = nullptr;
 236   GPR_ASSERT(grpc_compression_algorithm_name(algorithm_to_disable, &algo_name));
 237   std::string expected_details =
 238       absl::StrCat("Compression algorithm '", algo_name, "' is disabled.");
 239   /* and we expect a specific reason for it */
 240   GPR_ASSERT(0 == grpc_slice_str_cmp(details, expected_details.c_str()));
 241   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
 242
 243   grpc_slice_unref(details);
 244   grpc_metadata_array_destroy(&initial_metadata_recv);
 245   grpc_metadata_array_destroy(&trailing_metadata_recv);
 246   grpc_metadata_array_destroy(&request_metadata_recv);
 247   grpc_call_details_destroy(&call_details);
 248
 249   grpc_call_unref(c);
 250   grpc_call_unref(s);
 251
 252   cq_verifier_destroy(cqv);
 253
 254   grpc_slice_unref(request_payload_slice);
 255   grpc_byte_buffer_destroy(request_payload);
 256   grpc_byte_buffer_destroy(request_payload_recv);
 257
 258   {
 259     grpc_core::ExecCtx exec_ctx;
 260     grpc_channel_args_destroy(client_args);
 261     grpc_channel_args_destroy(server_args);
 262   }
 263
 264   end_test(&f);
 265   config.tear_down_data(&f);
 266 }
 267
 268 static void request_with_payload_template(
 269     grpc_end2end_test_config config, const char* test_name,
 270     uint32_t client_send_flags_bitmask,
 271     grpc_compression_algorithm default_client_channel_compression_algorithm,
 272     grpc_compression_algorithm default_server_channel_compression_algorithm,
 273     grpc_compression_algorithm /*expected_client_compression_algorithm*/,
 274     grpc_compression_algorithm /*expected_server_compression_algorithm*/,
 275     grpc_metadata* client_init_metadata, bool set_server_level,
 276     grpc_compression_level server_compression_level,
 277     bool send_message_before_initial_metadata,
 278     bool set_default_server_message_compression_algorithm,
 279     grpc_compression_algorithm default_server_message_compression_algorithm) {
 280   grpc_call* c;
 281   grpc_call* s;
 282   grpc_slice request_payload_slice;
 283   grpc_byte_buffer* request_payload = nullptr;
 284   grpc_channel_args* client_args;
 285   grpc_channel_args* server_args;
 286   grpc_end2end_test_fixture f;
 287   grpc_op ops[6];
 288   grpc_op* op;
 289   grpc_metadata_array initial_metadata_recv;
 290   grpc_metadata_array trailing_metadata_recv;
 291   grpc_metadata_array request_metadata_recv;
 292   grpc_byte_buffer* request_payload_recv = nullptr;
 293   grpc_byte_buffer* response_payload;
 294   grpc_byte_buffer* response_payload_recv;
 295   grpc_call_details call_details;
 296   grpc_status_code status;
 297   grpc_call_error error;
 298   grpc_slice details;
 299   int was_cancelled = 2;
 300   cq_verifier* cqv;
 301   char request_str[1024];
 302   char response_str[1024];
 303
 304   memset(request_str, 'x', 1023);
 305   request_str[1023] = '\0';
 306
 307   memset(response_str, 'y', 1023);
 308   response_str[1023] = '\0';
 309
 310   request_payload_slice = grpc_slice_from_copied_string(request_str);
 311   grpc_slice response_payload_slice =
 312       grpc_slice_from_copied_string(response_str);
 313
 314   client_args = grpc_channel_args_set_channel_default_compression_algorithm(
 315       nullptr, default_client_channel_compression_algorithm);
 316   if (set_default_server_message_compression_algorithm) {
 317     server_args = grpc_channel_args_set_channel_default_compression_algorithm(
 318         nullptr, default_server_message_compression_algorithm);
 319   } else {
 320     server_args = grpc_channel_args_set_channel_default_compression_algorithm(
 321         nullptr, default_server_channel_compression_algorithm);
 322   }
 323
 324   f = begin_test(config, test_name, client_args, server_args);
 325   cqv = cq_verifier_create(f.cq);
 326
 327   gpr_timespec deadline = five_seconds_from_now();
 328   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 329                                grpc_slice_from_static_string("/foo"), nullptr,
 330                                deadline, nullptr);
 331   GPR_ASSERT(c);
 332
 333   grpc_metadata_array_init(&initial_metadata_recv);
 334   grpc_metadata_array_init(&trailing_metadata_recv);
 335   grpc_metadata_array_init(&request_metadata_recv);
 336   grpc_call_details_init(&call_details);
 337
 338   if (send_message_before_initial_metadata) {
 339     request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 340     memset(ops, 0, sizeof(ops));
 341     op = ops;
 342     op->op = GRPC_OP_SEND_MESSAGE;
 343     op->data.send_message.send_message = request_payload;
 344     op->flags = client_send_flags_bitmask;
 345     op->reserved = nullptr;
 346     op++;
 347     error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(2),
 348                                   nullptr);
 349     GPR_ASSERT(GRPC_CALL_OK == error);
 350     CQ_EXPECT_COMPLETION(cqv, tag(2), true);
 351   }
 352
 353   memset(ops, 0, sizeof(ops));
 354   op = ops;
 355   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 356   if (client_init_metadata != nullptr) {
 357     op->data.send_initial_metadata.count = 1;
 358     op->data.send_initial_metadata.metadata = client_init_metadata;
 359   } else {
 360     op->data.send_initial_metadata.count = 0;
 361   }
 362   op->flags = 0;
 363   op->reserved = nullptr;
 364   op++;
 365   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 366   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 367   op->flags = 0;
 368   op->reserved = nullptr;
 369   op++;
 370   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 371   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 372   op->data.recv_status_on_client.status = &status;
 373   op->data.recv_status_on_client.status_details = &details;
 374   op->flags = 0;
 375   op->reserved = nullptr;
 376   op++;
 377   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 378                                 nullptr);
 379   GPR_ASSERT(GRPC_CALL_OK == error);
 380
 381   error =
 382       grpc_server_request_call(f.server, &s, &call_details,
 383                                &request_metadata_recv, f.cq, f.cq, tag(100));
 384   GPR_ASSERT(GRPC_CALL_OK == error);
 385   CQ_EXPECT_COMPLETION(cqv, tag(100), true);
 386   cq_verify(cqv);
 387
 388   GPR_ASSERT(GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(
 389                  s)) == GRPC_COMPRESS_ALGORITHMS_COUNT);
 390   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
 391                         GRPC_COMPRESS_NONE) != 0);
 392   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
 393                         GRPC_COMPRESS_DEFLATE) != 0);
 394   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
 395                         GRPC_COMPRESS_GZIP) != 0);
 396   GPR_ASSERT(GPR_BITGET(grpc_call_test_only_get_encodings_accepted_by_peer(s),
 397                         GRPC_COMPRESS_STREAM_GZIP) != 0);
 398   GPR_ASSERT(GPR_BITCOUNT(grpc_call_test_only_get_encodings_accepted_by_peer(
 399                  s)) == GRPC_COMPRESS_ALGORITHMS_COUNT);
 400
 401   memset(ops, 0, sizeof(ops));
 402   op = ops;
 403   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 404   op->data.send_initial_metadata.count = 0;
 405   if (set_server_level) {
 406     op->data.send_initial_metadata.maybe_compression_level.is_set = true;
 407     op->data.send_initial_metadata.maybe_compression_level.level =
 408         server_compression_level;
 409   }
 410   op->flags = 0;
 411   op->reserved = nullptr;
 412   op++;
 413   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 414   op->data.recv_close_on_server.cancelled = &was_cancelled;
 415   op->flags = 0;
 416   op->reserved = nullptr;
 417   op++;
 418   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(101),
 419                                 nullptr);
 420   GPR_ASSERT(GRPC_CALL_OK == error);
 421
 422   for (int i = 0; i < 2; i++) {
 423     response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
 424
 425     if (i > 0 || !send_message_before_initial_metadata) {
 426       request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 427       memset(ops, 0, sizeof(ops));
 428       op = ops;
 429       op->op = GRPC_OP_SEND_MESSAGE;
 430       op->data.send_message.send_message = request_payload;
 431       op->flags = client_send_flags_bitmask;
 432       op->reserved = nullptr;
 433       op++;
 434       error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops),
 435                                     tag(2), nullptr);
 436       GPR_ASSERT(GRPC_CALL_OK == error);
 437       CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
 438     }
 439
 440     memset(ops, 0, sizeof(ops));
 441     op = ops;
 442     op->op = GRPC_OP_RECV_MESSAGE;
 443     op->data.recv_message.recv_message = &request_payload_recv;
 444     op->flags = 0;
 445     op->reserved = nullptr;
 446     op++;
 447     error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
 448                                   tag(102), nullptr);
 449     GPR_ASSERT(GRPC_CALL_OK == error);
 450
 451     CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
 452     cq_verify(cqv);
 453
 454     GPR_ASSERT(request_payload_recv->type == GRPC_BB_RAW);
 455     GPR_ASSERT(byte_buffer_eq_string(request_payload_recv, request_str));
 456
 457     memset(ops, 0, sizeof(ops));
 458     op = ops;
 459     op->op = GRPC_OP_SEND_MESSAGE;
 460     op->data.send_message.send_message = response_payload;
 461     op->flags = 0;
 462     op->reserved = nullptr;
 463     op++;
 464     error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
 465                                   tag(103), nullptr);
 466     GPR_ASSERT(GRPC_CALL_OK == error);
 467
 468     memset(ops, 0, sizeof(ops));
 469     op = ops;
 470     op->op = GRPC_OP_RECV_MESSAGE;
 471     op->data.recv_message.recv_message = &response_payload_recv;
 472     op->flags = 0;
 473     op->reserved = nullptr;
 474     op++;
 475     error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(3),
 476                                   nullptr);
 477     GPR_ASSERT(GRPC_CALL_OK == error);
 478
 479     CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
 480     CQ_EXPECT_COMPLETION(cqv, tag(3), 1);
 481     cq_verify(cqv);
 482
 483     GPR_ASSERT(response_payload_recv->type == GRPC_BB_RAW);
 484     GPR_ASSERT(byte_buffer_eq_string(response_payload_recv, response_str));
 485
 486     grpc_byte_buffer_destroy(request_payload);
 487     grpc_byte_buffer_destroy(response_payload);
 488     grpc_byte_buffer_destroy(request_payload_recv);
 489     grpc_byte_buffer_destroy(response_payload_recv);
 490   }
 491
 492   grpc_slice_unref(request_payload_slice);
 493   grpc_slice_unref(response_payload_slice);
 494
 495   memset(ops, 0, sizeof(ops));
 496   op = ops;
 497   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 498   op->flags = 0;
 499   op->reserved = nullptr;
 500   op++;
 501   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(4),
 502                                 nullptr);
 503   GPR_ASSERT(GRPC_CALL_OK == error);
 504
 505   memset(ops, 0, sizeof(ops));
 506   op = ops;
 507   op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
 508   op->data.send_status_from_server.trailing_metadata_count = 0;
 509   op->data.send_status_from_server.status = GRPC_STATUS_OK;
 510   grpc_slice status_details = grpc_slice_from_static_string("xyz");
 511   op->data.send_status_from_server.status_details = &status_details;
 512   op->flags = 0;
 513   op->reserved = nullptr;
 514   op++;
 515   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(104),
 516                                 nullptr);
 517   GPR_ASSERT(GRPC_CALL_OK == error);
 518
 519   CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
 520   CQ_EXPECT_COMPLETION(cqv, tag(4), 1);
 521   CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
 522   CQ_EXPECT_COMPLETION(cqv, tag(104), 1);
 523   cq_verify(cqv);
 524
 525   GPR_ASSERT(status == GRPC_STATUS_OK);
 526   GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
 527   GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
 528   GPR_ASSERT(was_cancelled == 0);
 529
 530   grpc_slice_unref(details);
 531   grpc_metadata_array_destroy(&initial_metadata_recv);
 532   grpc_metadata_array_destroy(&trailing_metadata_recv);
 533   grpc_metadata_array_destroy(&request_metadata_recv);
 534   grpc_call_details_destroy(&call_details);
 535
 536   grpc_call_unref(c);
 537   grpc_call_unref(s);
 538
 539   cq_verifier_destroy(cqv);
 540
 541   {
 542     grpc_core::ExecCtx exec_ctx;
 543     grpc_channel_args_destroy(client_args);
 544     grpc_channel_args_destroy(server_args);
 545   }
 546
 547   end_test(&f);
 548   config.tear_down_data(&f);
 549 }
 550
 551 static void test_invoke_request_with_compressed_payload(
 552     grpc_end2end_test_config config) {
 553   request_with_payload_template(
 554       config, "test_invoke_request_with_compressed_payload", 0,
 555       GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_STREAM_GZIP,
 556       GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_STREAM_GZIP, nullptr,
 557       false, /* ignored */
 558       GRPC_COMPRESS_LEVEL_NONE, false, false, GRPC_COMPRESS_NONE);
 559 }
 560
 561 static void test_invoke_request_with_send_message_before_initial_metadata(
 562     grpc_end2end_test_config config) {
 563   request_with_payload_template(
 564       config, "test_invoke_request_with_send_message_before_initial_metadata",
 565       0, GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_STREAM_GZIP,
 566       GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_STREAM_GZIP, nullptr,
 567       false, /* ignored */
 568       GRPC_COMPRESS_LEVEL_NONE, true, false, GRPC_COMPRESS_NONE);
 569 }
 570
 571 static void test_invoke_request_with_compressed_payload_md_override(
 572     grpc_end2end_test_config config) {
 573   grpc_metadata gzip_compression_override;
 574   grpc_metadata identity_compression_override;
 575
 576   gzip_compression_override.key =
 577       GRPC_MDSTR_GRPC_INTERNAL_STREAM_ENCODING_REQUEST;
 578   gzip_compression_override.value =
 579       grpc_slice_from_static_string("stream/gzip");
 580   memset(&gzip_compression_override.internal_data, 0,
 581          sizeof(gzip_compression_override.internal_data));
 582
 583   identity_compression_override.key =
 584       GRPC_MDSTR_GRPC_INTERNAL_STREAM_ENCODING_REQUEST;
 585   identity_compression_override.value =
 586       grpc_slice_from_static_string("identity");
 587   memset(&identity_compression_override.internal_data, 0,
 588          sizeof(identity_compression_override.internal_data));
 589
 590   /* Channel default NONE (aka IDENTITY), call override to stream GZIP */
 591   request_with_payload_template(
 592       config, "test_invoke_request_with_compressed_payload_md_override_1", 0,
 593       GRPC_COMPRESS_NONE, GRPC_COMPRESS_NONE, GRPC_COMPRESS_STREAM_GZIP,
 594       GRPC_COMPRESS_NONE, &gzip_compression_override, false,
 595       /*ignored*/ GRPC_COMPRESS_LEVEL_NONE, false, false, GRPC_COMPRESS_NONE);
 596
 597   /* Channel default stream GZIP, call override to NONE (aka IDENTITY) */
 598   request_with_payload_template(
 599       config, "test_invoke_request_with_compressed_payload_md_override_3", 0,
 600       GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_NONE, GRPC_COMPRESS_NONE,
 601       GRPC_COMPRESS_NONE, &identity_compression_override, false,
 602       /*ignored*/ GRPC_COMPRESS_LEVEL_NONE, false, false, GRPC_COMPRESS_NONE);
 603 }
 604
 605 static void test_invoke_request_with_disabled_algorithm(
 606     grpc_end2end_test_config config) {
 607   request_for_disabled_algorithm(
 608       config, "test_invoke_request_with_disabled_algorithm", 0,
 609       GRPC_COMPRESS_STREAM_GZIP, GRPC_COMPRESS_STREAM_GZIP,
 610       GRPC_STATUS_UNIMPLEMENTED, nullptr);
 611 }
 612
 613 void stream_compression_compressed_payload(grpc_end2end_test_config config) {
 614   test_invoke_request_with_compressed_payload(config);
 615   test_invoke_request_with_send_message_before_initial_metadata(config);
 616   test_invoke_request_with_compressed_payload_md_override(config);
 617   test_invoke_request_with_disabled_algorithm(config);
 618 }
 619
 620 void stream_compression_compressed_payload_pre_init(void) {}