test/core/end2end/tests/stream_compression_ping_pong_streaming.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 <grpc/byte_buffer.h>
  25 #include <grpc/compression.h>
  26 #include <grpc/support/alloc.h>
  27 #include <grpc/support/log.h>
  28 #include <grpc/support/time.h>
  29
  30 #include "src/core/lib/channel/channel_args.h"
  31 #include "src/core/lib/compression/compression_args.h"
  32 #include "src/core/lib/surface/call.h"
  33 #include "test/core/end2end/cq_verifier.h"
  34
  35 static void* tag(intptr_t t) { return (void*)t; }
  36
  37 static grpc_end2end_test_fixture begin_test(grpc_end2end_test_config config,
  38                                             const char* test_name,
  39                                             grpc_channel_args* client_args,
  40                                             grpc_channel_args* server_args) {
  41   grpc_end2end_test_fixture f;
  42   gpr_log(GPR_INFO, "Running test: %s/%s", test_name, config.name);
  43   f = config.create_fixture(client_args, server_args);
  44   config.init_server(&f, server_args);
  45   config.init_client(&f, client_args);
  46   return f;
  47 }
  48
  49 static gpr_timespec n_seconds_from_now(int n) {
  50   return grpc_timeout_seconds_to_deadline(n);
  51 }
  52
  53 static gpr_timespec five_seconds_from_now(void) {
  54   return n_seconds_from_now(5);
  55 }
  56
  57 static void drain_cq(grpc_completion_queue* cq) {
  58   grpc_event ev;
  59   do {
  60     ev = grpc_completion_queue_next(cq, five_seconds_from_now(), nullptr);
  61   } while (ev.type != GRPC_QUEUE_SHUTDOWN);
  62 }
  63
  64 static void shutdown_server(grpc_end2end_test_fixture* f) {
  65   if (!f->server) return;
  66   grpc_server_shutdown_and_notify(f->server, f->shutdown_cq, tag(1000));
  67   GPR_ASSERT(grpc_completion_queue_pluck(f->shutdown_cq, tag(1000),
  68                                          grpc_timeout_seconds_to_deadline(5),
  69                                          nullptr)
  70                  .type == GRPC_OP_COMPLETE);
  71   grpc_server_destroy(f->server);
  72   f->server = nullptr;
  73 }
  74
  75 static void shutdown_client(grpc_end2end_test_fixture* f) {
  76   if (!f->client) return;
  77   grpc_channel_destroy(f->client);
  78   f->client = nullptr;
  79 }
  80
  81 static void end_test(grpc_end2end_test_fixture* f) {
  82   shutdown_server(f);
  83   shutdown_client(f);
  84
  85   grpc_completion_queue_shutdown(f->cq);
  86   drain_cq(f->cq);
  87   grpc_completion_queue_destroy(f->cq);
  88   grpc_completion_queue_destroy(f->shutdown_cq);
  89 }
  90
  91 /* Client pings and server pongs. Repeat messages rounds before finishing. */
  92 static void test_pingpong_streaming(grpc_end2end_test_config config,
  93                                     int messages) {
  94   grpc_channel_args* client_args = grpc_channel_args_set_compression_algorithm(
  95       nullptr, GRPC_COMPRESS_STREAM_GZIP);
  96   grpc_channel_args* server_args = grpc_channel_args_set_compression_algorithm(
  97       nullptr, GRPC_COMPRESS_STREAM_GZIP);
  98   grpc_end2end_test_fixture f =
  99       begin_test(config, "test_pingpong_streaming", client_args, server_args);
 100   grpc_call* c;
 101   grpc_call* s;
 102   cq_verifier* cqv = cq_verifier_create(f.cq);
 103   grpc_op ops[6];
 104   grpc_op* op;
 105   grpc_metadata_array initial_metadata_recv;
 106   grpc_metadata_array trailing_metadata_recv;
 107   grpc_metadata_array request_metadata_recv;
 108   grpc_call_details call_details;
 109   grpc_status_code status;
 110   grpc_call_error error;
 111   grpc_slice details;
 112   int was_cancelled = 2;
 113   grpc_byte_buffer* request_payload;
 114   grpc_byte_buffer* request_payload_recv;
 115   grpc_byte_buffer* response_payload;
 116   grpc_byte_buffer* response_payload_recv;
 117   int i;
 118   grpc_slice request_payload_slice =
 119       grpc_slice_from_copied_string("hello world");
 120   grpc_slice response_payload_slice =
 121       grpc_slice_from_copied_string("hello you");
 122
 123   gpr_timespec deadline = five_seconds_from_now();
 124   c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
 125                                grpc_slice_from_static_string("/foo"), nullptr,
 126                                deadline, nullptr);
 127   GPR_ASSERT(c);
 128
 129   grpc_metadata_array_init(&initial_metadata_recv);
 130   grpc_metadata_array_init(&trailing_metadata_recv);
 131   grpc_metadata_array_init(&request_metadata_recv);
 132   grpc_call_details_init(&call_details);
 133
 134   memset(ops, 0, sizeof(ops));
 135   op = ops;
 136   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 137   op->data.send_initial_metadata.count = 0;
 138   op->flags = 0;
 139   op->reserved = nullptr;
 140   op++;
 141   op->op = GRPC_OP_RECV_INITIAL_METADATA;
 142   op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
 143   op->flags = 0;
 144   op->reserved = nullptr;
 145   op++;
 146   op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
 147   op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
 148   op->data.recv_status_on_client.status = &status;
 149   op->data.recv_status_on_client.status_details = &details;
 150   op->flags = 0;
 151   op->reserved = nullptr;
 152   op++;
 153   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
 154                                 nullptr);
 155   GPR_ASSERT(GRPC_CALL_OK == error);
 156
 157   error =
 158       grpc_server_request_call(f.server, &s, &call_details,
 159                                &request_metadata_recv, f.cq, f.cq, tag(100));
 160   GPR_ASSERT(GRPC_CALL_OK == error);
 161   CQ_EXPECT_COMPLETION(cqv, tag(100), 1);
 162   cq_verify(cqv);
 163
 164   memset(ops, 0, sizeof(ops));
 165   op = ops;
 166   op->op = GRPC_OP_SEND_INITIAL_METADATA;
 167   op->data.send_initial_metadata.count = 0;
 168   op->flags = 0;
 169   op->reserved = nullptr;
 170   op++;
 171   op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
 172   op->data.recv_close_on_server.cancelled = &was_cancelled;
 173   op->flags = 0;
 174   op->reserved = nullptr;
 175   op++;
 176   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(101),
 177                                 nullptr);
 178   GPR_ASSERT(GRPC_CALL_OK == error);
 179
 180   for (i = 0; i < messages; i++) {
 181     request_payload = grpc_raw_byte_buffer_create(&request_payload_slice, 1);
 182     response_payload = grpc_raw_byte_buffer_create(&response_payload_slice, 1);
 183
 184     memset(ops, 0, sizeof(ops));
 185     op = ops;
 186     op->op = GRPC_OP_SEND_MESSAGE;
 187     op->data.send_message.send_message = request_payload;
 188     op->flags = 0;
 189     op->reserved = nullptr;
 190     op++;
 191     op->op = GRPC_OP_RECV_MESSAGE;
 192     op->data.recv_message.recv_message = &response_payload_recv;
 193     op->flags = 0;
 194     op->reserved = nullptr;
 195     op++;
 196     error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(2),
 197                                   nullptr);
 198     GPR_ASSERT(GRPC_CALL_OK == error);
 199
 200     memset(ops, 0, sizeof(ops));
 201     op = ops;
 202     op->op = GRPC_OP_RECV_MESSAGE;
 203     op->data.recv_message.recv_message = &request_payload_recv;
 204     op->flags = 0;
 205     op->reserved = nullptr;
 206     op++;
 207     error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
 208                                   tag(102), nullptr);
 209     GPR_ASSERT(GRPC_CALL_OK == error);
 210     CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
 211     cq_verify(cqv);
 212
 213     memset(ops, 0, sizeof(ops));
 214     op = ops;
 215     op->op = GRPC_OP_SEND_MESSAGE;
 216     op->data.send_message.send_message = response_payload;
 217     op->flags = 0;
 218     op->reserved = nullptr;
 219     op++;
 220     error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops),
 221                                   tag(103), nullptr);
 222     GPR_ASSERT(GRPC_CALL_OK == error);
 223     CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
 224     CQ_EXPECT_COMPLETION(cqv, tag(2), 1);
 225     cq_verify(cqv);
 226
 227     grpc_byte_buffer_destroy(request_payload);
 228     grpc_byte_buffer_destroy(response_payload);
 229     grpc_byte_buffer_destroy(request_payload_recv);
 230     grpc_byte_buffer_destroy(response_payload_recv);
 231   }
 232
 233   grpc_slice_unref(request_payload_slice);
 234   grpc_slice_unref(response_payload_slice);
 235
 236   memset(ops, 0, sizeof(ops));
 237   op = ops;
 238   op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
 239   op->flags = 0;
 240   op->reserved = nullptr;
 241   op++;
 242   error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(3),
 243                                 nullptr);
 244   GPR_ASSERT(GRPC_CALL_OK == error);
 245
 246   memset(ops, 0, sizeof(ops));
 247   op = ops;
 248   op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
 249   op->data.send_status_from_server.trailing_metadata_count = 0;
 250   op->data.send_status_from_server.status = GRPC_STATUS_UNIMPLEMENTED;
 251   grpc_slice status_details = grpc_slice_from_static_string("xyz");
 252   op->data.send_status_from_server.status_details = &status_details;
 253   op->flags = 0;
 254   op->reserved = nullptr;
 255   op++;
 256   error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(104),
 257                                 nullptr);
 258   GPR_ASSERT(GRPC_CALL_OK == error);
 259
 260   CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
 261   CQ_EXPECT_COMPLETION(cqv, tag(3), 1);
 262   CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
 263   CQ_EXPECT_COMPLETION(cqv, tag(104), 1);
 264   cq_verify(cqv);
 265
 266   grpc_call_unref(c);
 267   grpc_call_unref(s);
 268
 269   cq_verifier_destroy(cqv);
 270
 271   grpc_metadata_array_destroy(&initial_metadata_recv);
 272   grpc_metadata_array_destroy(&trailing_metadata_recv);
 273   grpc_metadata_array_destroy(&request_metadata_recv);
 274   grpc_call_details_destroy(&call_details);
 275   grpc_slice_unref(details);
 276
 277   end_test(&f);
 278   config.tear_down_data(&f);
 279   {
 280     grpc_core::ExecCtx exec_ctx;
 281     grpc_channel_args_destroy(client_args);
 282     grpc_channel_args_destroy(server_args);
 283   }
 284 }
 285
 286 void stream_compression_ping_pong_streaming(grpc_end2end_test_config config) {
 287   int i;
 288
 289   for (i = 1; i < 10; i++) {
 290     test_pingpong_streaming(config, i);
 291   }
 292 }
 293
 294 void stream_compression_ping_pong_streaming_pre_init(void) {}