Line data    Source code

       1             : /*
       2             :  *
       3             :  * Copyright 2015, Google Inc.
       4             :  * All rights reserved.
       5             :  *
       6             :  * Redistribution and use in source and binary forms, with or without
       7             :  * modification, are permitted provided that the following conditions are
       8             :  * met:
       9             :  *
      10             :  *     * Redistributions of source code must retain the above copyright
      11             :  * notice, this list of conditions and the following disclaimer.
      12             :  *     * Redistributions in binary form must reproduce the above
      13             :  * copyright notice, this list of conditions and the following disclaimer
      14             :  * in the documentation and/or other materials provided with the
      15             :  * distribution.
      16             :  *     * Neither the name of Google Inc. nor the names of its
      17             :  * contributors may be used to endorse or promote products derived from
      18             :  * this software without specific prior written permission.
      19             :  *
      20             :  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
      21             :  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
      22             :  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
      23             :  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
      24             :  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
      25             :  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
      26             :  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
      27             :  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
      28             :  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
      29             :  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
      30             :  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
      31             :  *
      32             :  */
      33             : 
      34             : #ifndef GRPCXX_SUPPORT_ASYNC_UNARY_CALL_H
      35             : #define GRPCXX_SUPPORT_ASYNC_UNARY_CALL_H
      36             : 
      37             : #include <grpc/support/log.h>
      38             : #include <grpc++/channel.h>
      39             : #include <grpc++/client_context.h>
      40             : #include <grpc++/completion_queue.h>
      41             : #include <grpc++/server_context.h>
      42             : #include <grpc++/impl/call.h>
      43             : #include <grpc++/impl/service_type.h>
      44             : #include <grpc++/support/status.h>
      45             : 
      46             : namespace grpc {
      47             : 
      48             : template <class R>
      49     1139138 : class ClientAsyncResponseReaderInterface {
      50             :  public:
      51     1138769 :   virtual ~ClientAsyncResponseReaderInterface() {}
      52             :   virtual void ReadInitialMetadata(void* tag) = 0;
      53             :   virtual void Finish(R* msg, Status* status, void* tag) = 0;
      54             : };
      55             : 
      56             : template <class R>
      57     2278052 : class ClientAsyncResponseReader GRPC_FINAL
      58             :     : public ClientAsyncResponseReaderInterface<R> {
      59             :  public:
      60             :   template <class W>
      61     1139473 :   ClientAsyncResponseReader(Channel* channel, CompletionQueue* cq,
      62             :                             const RpcMethod& method, ClientContext* context,
      63             :                             const W& request)
      64     1139473 :       : context_(context), call_(channel->CreateCall(method, context, cq)) {
      65     1139326 :     init_buf_.SendInitialMetadata(context->send_initial_metadata_);
      66             :     // TODO(ctiller): don't assert
      67     1139540 :     GPR_ASSERT(init_buf_.SendMessage(request).ok());
      68     1139449 :     init_buf_.ClientSendClose();
      69     1139452 :     call_.PerformOps(&init_buf_);
      70     1139538 :   }
      71             : 
      72           4 :   void ReadInitialMetadata(void* tag) {
      73           4 :     GPR_ASSERT(!context_->initial_metadata_received_);
      74             : 
      75           4 :     meta_buf_.set_output_tag(tag);
      76           4 :     meta_buf_.RecvInitialMetadata(context_);
      77           4 :     call_.PerformOps(&meta_buf_);
      78           4 :   }
      79             : 
      80     1139050 :   void Finish(R* msg, Status* status, void* tag) {
      81     1139050 :     finish_buf_.set_output_tag(tag);
      82     1139620 :     if (!context_->initial_metadata_received_) {
      83     1139568 :       finish_buf_.RecvInitialMetadata(context_);
      84             :     }
      85     1139654 :     finish_buf_.RecvMessage(msg);
      86     1139561 :     finish_buf_.ClientRecvStatus(context_, status);
      87     1139597 :     call_.PerformOps(&finish_buf_);
      88     1138781 :   }
      89             : 
      90             :  private:
      91             :   ClientContext* context_;
      92             :   Call call_;
      93             :   SneakyCallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
      94             :                   CallOpClientSendClose> init_buf_;
      95             :   CallOpSet<CallOpRecvInitialMetadata> meta_buf_;
      96             :   CallOpSet<CallOpRecvInitialMetadata, CallOpRecvMessage<R>,
      97             :             CallOpClientRecvStatus> finish_buf_;
      98             : };
      99             : 
     100             : template <class W>
     101     2320356 : class ServerAsyncResponseWriter GRPC_FINAL
     102             :     : public ServerAsyncStreamingInterface {
     103             :  public:
     104     1180633 :   explicit ServerAsyncResponseWriter(ServerContext* ctx)
     105     1180633 :       : call_(nullptr, nullptr, nullptr), ctx_(ctx) {}
     106             : 
     107           6 :   void SendInitialMetadata(void* tag) GRPC_OVERRIDE {
     108           6 :     GPR_ASSERT(!ctx_->sent_initial_metadata_);
     109             : 
     110           6 :     meta_buf_.set_output_tag(tag);
     111           6 :     meta_buf_.SendInitialMetadata(ctx_->initial_metadata_);
     112           6 :     ctx_->sent_initial_metadata_ = true;
     113           6 :     call_.PerformOps(&meta_buf_);
     114           6 :   }
     115             : 
     116     1139544 :   void Finish(const W& msg, const Status& status, void* tag) {
     117     1139544 :     finish_buf_.set_output_tag(tag);
     118     1139560 :     if (!ctx_->sent_initial_metadata_) {
     119     1139552 :       finish_buf_.SendInitialMetadata(ctx_->initial_metadata_);
     120     1139566 :       ctx_->sent_initial_metadata_ = true;
     121             :     }
     122             :     // The response is dropped if the status is not OK.
     123     1139574 :     if (status.ok()) {
     124     1139571 :       finish_buf_.ServerSendStatus(ctx_->trailing_metadata_,
     125             :                                    finish_buf_.SendMessage(msg));
     126             :     } else {
     127           0 :       finish_buf_.ServerSendStatus(ctx_->trailing_metadata_, status);
     128             :     }
     129     1139563 :     call_.PerformOps(&finish_buf_);
     130     1139410 :   }
     131             : 
     132             :   void FinishWithError(const Status& status, void* tag) {
     133             :     GPR_ASSERT(!status.ok());
     134             :     finish_buf_.set_output_tag(tag);
     135             :     if (!ctx_->sent_initial_metadata_) {
     136             :       finish_buf_.SendInitialMetadata(ctx_->initial_metadata_);
     137             :       ctx_->sent_initial_metadata_ = true;
     138             :     }
     139             :     finish_buf_.ServerSendStatus(ctx_->trailing_metadata_, status);
     140             :     call_.PerformOps(&finish_buf_);
     141             :   }
     142             : 
     143             :  private:
     144     1180676 :   void BindCall(Call* call) GRPC_OVERRIDE { call_ = *call; }
     145             : 
     146             :   Call call_;
     147             :   ServerContext* ctx_;
     148             :   CallOpSet<CallOpSendInitialMetadata> meta_buf_;
     149             :   CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
     150             :             CallOpServerSendStatus> finish_buf_;
     151             : };
     152             : 
     153             : }  // namespace grpc
     154             : 
     155             : #endif  // GRPCXX_SUPPORT_ASYNC_UNARY_CALL_H