third_party/pigweed/repo/pw_rpc/base_server_writer.cc

   1 // Copyright 2020 The Pigweed Authors
   2 //
   3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
   4 // use this file except in compliance with the License. You may obtain a copy of
   5 // the License at
   6 //
   7 //     https://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 // Unless required by applicable law or agreed to in writing, software
  10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 // License for the specific language governing permissions and limitations under
  13 // the License.
  14
  15 #include "pw_rpc/internal/base_server_writer.h"
  16
  17 #include "pw_rpc/internal/method.h"
  18 #include "pw_rpc/internal/packet.h"
  19 #include "pw_rpc/internal/server.h"
  20
  21 namespace pw::rpc::internal {
  22
  23 BaseServerWriter::BaseServerWriter(ServerCall& call)
  24     : call_(call), state_(kOpen) {
  25   call_.server().RegisterWriter(*this);
  26 }
  27
  28 BaseServerWriter& BaseServerWriter::operator=(BaseServerWriter&& other) {
  29   Finish();
  30
  31   state_ = other.state_;
  32
  33   if (other.open()) {
  34     other.call_.server().RemoveWriter(other);
  35     other.state_ = kClosed;
  36
  37     other.call_.server().RegisterWriter(*this);
  38   }
  39
  40   call_ = std::move(other.call_);
  41   response_ = std::move(other.response_);
  42
  43   return *this;
  44 }
  45
  46 uint32_t BaseServerWriter::method_id() const { return call_.method().id(); }
  47
  48 void BaseServerWriter::Finish(Status status) {
  49   if (!open()) {
  50     return;
  51   }
  52
  53   // If the ServerWriter implementer or user forgets to release an acquired
  54   // buffer before finishing, release it here.
  55   if (!response_.empty()) {
  56     ReleasePayloadBuffer();
  57   }
  58
  59   Close();
  60
  61   // Send a control packet indicating that the stream (and RPC) has terminated.
  62   call_.channel().Send(Packet(PacketType::SERVER_STREAM_END,
  63                               call_.channel().id(),
  64                               call_.service().id(),
  65                               method().id(),
  66                               {},
  67                               status));
  68 }
  69
  70 std::span<std::byte> BaseServerWriter::AcquirePayloadBuffer() {
  71   if (!open()) {
  72     return {};
  73   }
  74
  75   // Only allow having one active buffer at a time.
  76   if (response_.empty()) {
  77     response_ = call_.channel().AcquireBuffer();
  78   }
  79
  80   return response_.payload(ResponsePacket());
  81 }
  82
  83 Status BaseServerWriter::ReleasePayloadBuffer(
  84     std::span<const std::byte> payload) {
  85   if (!open()) {
  86     return Status::FailedPrecondition();
  87   }
  88   return call_.channel().Send(response_, ResponsePacket(payload));
  89 }
  90
  91 Status BaseServerWriter::ReleasePayloadBuffer() {
  92   if (!open()) {
  93     return Status::FailedPrecondition();
  94   }
  95
  96   call_.channel().Release(response_);
  97   return OkStatus();
  98 }
  99
 100 void BaseServerWriter::Close() {
 101   if (!open()) {
 102     return;
 103   }
 104
 105   call_.server().RemoveWriter(*this);
 106   state_ = kClosed;
 107 }
 108
 109 Packet BaseServerWriter::ResponsePacket(
 110     std::span<const std::byte> payload) const {
 111   return Packet(PacketType::RESPONSE,
 112                 call_.channel().id(),
 113                 call_.service().id(),
 114                 method().id(),
 115                 payload);
 116 }
 117
 118 }  // namespace pw::rpc::internal