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_assert/assert.h"
18 #include "pw_rpc/internal/method.h"
19 #include "pw_rpc/internal/packet.h"
20 #include "pw_rpc/internal/server.h"
21
22 namespace pw::rpc::internal {
23
BaseServerWriter(ServerCall & call)24 BaseServerWriter::BaseServerWriter(ServerCall& call)
25 : call_(call), state_(kOpen) {
26 call_.server().RegisterWriter(*this);
27 }
28
operator =(BaseServerWriter && other)29 BaseServerWriter& BaseServerWriter::operator=(BaseServerWriter&& other) {
30 Finish();
31
32 state_ = other.state_;
33
34 if (other.open()) {
35 other.call_.server().RemoveWriter(other);
36 other.state_ = kClosed;
37
38 other.call_.server().RegisterWriter(*this);
39 }
40
41 call_ = std::move(other.call_);
42 response_ = std::move(other.response_);
43
44 return *this;
45 }
46
method_id() const47 uint32_t BaseServerWriter::method_id() const { return call_.method().id(); }
48
Finish(Status status)49 Status BaseServerWriter::Finish(Status status) {
50 if (!open()) {
51 return Status::FailedPrecondition();
52 }
53
54 // If the ServerWriter implementer or user forgets to release an acquired
55 // buffer before finishing, release it here.
56 if (!response_.empty()) {
57 ReleasePayloadBuffer();
58 }
59
60 Close();
61
62 // Send a control packet indicating that the stream (and RPC) has terminated.
63 return call_.channel().Send(Packet(PacketType::SERVER_STREAM_END,
64 call_.channel().id(),
65 call_.service().id(),
66 method().id(),
67 {},
68 status));
69 }
70
AcquirePayloadBuffer()71 std::span<std::byte> BaseServerWriter::AcquirePayloadBuffer() {
72 PW_DCHECK(open());
73
74 // Only allow having one active buffer at a time.
75 if (response_.empty()) {
76 response_ = call_.channel().AcquireBuffer();
77 }
78
79 return response_.payload(ResponsePacket());
80 }
81
ReleasePayloadBuffer(std::span<const std::byte> payload)82 Status BaseServerWriter::ReleasePayloadBuffer(
83 std::span<const std::byte> payload) {
84 PW_DCHECK(open());
85 return call_.channel().Send(response_, ResponsePacket(payload));
86 }
87
ReleasePayloadBuffer()88 Status BaseServerWriter::ReleasePayloadBuffer() {
89 PW_DCHECK(open());
90 call_.channel().Release(response_);
91 return OkStatus();
92 }
93
Close()94 void BaseServerWriter::Close() {
95 if (!open()) {
96 return;
97 }
98
99 call_.server().RemoveWriter(*this);
100 state_ = kClosed;
101 }
102
ResponsePacket(std::span<const std::byte> payload) const103 Packet BaseServerWriter::ResponsePacket(
104 std::span<const std::byte> payload) const {
105 return Packet(PacketType::RESPONSE,
106 call_.channel().id(),
107 call_.service().id(),
108 method().id(),
109 payload);
110 }
111
112 } // namespace pw::rpc::internal
113