pw_rpc: Support asynchronous unary RPCs

- Implement asynchronous unary RPCs for the server. Define
  ServerResponder classes as the ServerWriter equivalent for unary RPCs.
- Use the responder classes instead of custom logic for synchronous
  unary RPCs.
- Expand tests to cover asynchronous unary RPCs.

Change-Id: I57e6fe26efbfd7c140f6b7c486ea846702967d2b
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/60400
Pigweed-Auto-Submit: Wyatt Hepler <hepler@google.com>
Commit-Queue: Wyatt Hepler <hepler@google.com>
Reviewed-by: Alexei Frolov <frolv@google.com>
diff --git a/pw_rpc/call.cc b/pw_rpc/call.cc
index b7cd614..083c106 100644
--- a/pw_rpc/call.cc
+++ b/pw_rpc/call.cc
@@ -16,23 +16,11 @@
 
 #include "pw_assert/check.h"
 #include "pw_rpc/internal/method.h"
-#include "pw_rpc/internal/packet.h"
 #include "pw_rpc/server.h"
 
 namespace pw::rpc::internal {
 namespace {
 
-Packet ResponsePacket(const CallContext& call,
-                      std::span<const std::byte> payload,
-                      Status status) {
-  return Packet(PacketType::RESPONSE,
-                call.channel().id(),
-                call.service().id(),
-                call.method().id(),
-                payload,
-                status);
-}
-
 Packet StreamPacket(const CallContext& call,
                     std::span<const std::byte> payload) {
   return Packet(PacketType::SERVER_STREAM,
@@ -83,8 +71,9 @@
 
 uint32_t Call::method_id() const { return call_.method().id(); }
 
-Status Call::CloseAndSendResponse(std::span<const std::byte> response,
-                                  Status status) {
+Status Call::CloseAndSendFinalPacket(PacketType type,
+                                     std::span<const std::byte> payload,
+                                     Status status) {
   if (!active()) {
     return Status::FailedPrecondition();
   }
@@ -98,8 +87,9 @@
 
   // Send a packet indicating that the RPC has terminated and optionally
   // containing the final payload.
-  packet_status =
-      call_.channel().Send(response_, ResponsePacket(call_, response, status));
+  packet_status = call_.channel().Send(
+      response_,
+      Packet(type, channel_id(), service_id(), method_id(), payload, status));
 
   Close();
 
diff --git a/pw_rpc/client_server_test.cc b/pw_rpc/client_server_test.cc
index b4c09cc..07ff7d0 100644
--- a/pw_rpc/client_server_test.cc
+++ b/pw_rpc/client_server_test.cc
@@ -40,7 +40,7 @@
   FakeService(uint32_t id) : Service(id, kMethods) {}
 
   static constexpr std::array<RawMethodUnion, 1> kMethods = {
-      RawMethod::Unary<FakeMethod>(kFakeMethodId),
+      RawMethod::SynchronousUnary<FakeMethod>(kFakeMethodId),
   };
 };
 
diff --git a/pw_rpc/nanopb/codegen_test.cc b/pw_rpc/nanopb/codegen_test.cc
index 2623243..df65ad6 100644
--- a/pw_rpc/nanopb/codegen_test.cc
+++ b/pw_rpc/nanopb/codegen_test.cc
@@ -31,6 +31,14 @@
     return static_cast<Status::Code>(request.status_code);
   }
 
+  void TestAnotherUnaryRpc(
+      ServerContext& ctx,
+      const pw_rpc_test_TestRequest& request,
+      NanopbServerResponder<pw_rpc_test_TestResponse>& responder) {
+    pw_rpc_test_TestResponse response{};
+    responder.Finish(response, TestUnaryRpc(ctx, request, response));
+  }
+
   static void TestServerStreamRpc(
       ServerContext&,
       const pw_rpc_test_TestRequest& request,
@@ -87,6 +95,20 @@
   EXPECT_EQ(1000, context.response().value);
 }
 
+TEST(NanopbCodegen, Server_InvokeAsyncUnaryRpc) {
+  PW_NANOPB_TEST_METHOD_CONTEXT(test::TestService, TestAnotherUnaryRpc) context;
+
+  context.call({.integer = 123, .status_code = OkStatus().code()});
+
+  EXPECT_EQ(OkStatus(), context.status());
+  EXPECT_EQ(124, context.response().value);
+
+  context.call(
+      {.integer = 999, .status_code = Status::InvalidArgument().code()});
+  EXPECT_EQ(Status::InvalidArgument(), context.status());
+  EXPECT_EQ(1000, context.response().value);
+}
+
 TEST(NanopbCodegen, Server_InvokeServerStreamingRpc) {
   PW_NANOPB_TEST_METHOD_CONTEXT(test::TestService, TestServerStreamRpc) context;
 
diff --git a/pw_rpc/nanopb/method.cc b/pw_rpc/nanopb/method.cc
index 129d627..a3e4014 100644
--- a/pw_rpc/nanopb/method.cc
+++ b/pw_rpc/nanopb/method.cc
@@ -33,9 +33,10 @@
     return;
   }
 
+  GenericNanopbResponder responder(call, MethodType::kUnary);
   const Status status =
       function_.synchronous_unary(call, request_struct, response_struct);
-  SendResponse(call.channel(), request, response_struct, status);
+  responder.SendResponse(response_struct, status).IgnoreError();
 }
 
 void NanopbMethod::CallUnaryRequest(CallContext& call,
@@ -63,41 +64,5 @@
   return false;
 }
 
-void NanopbMethod::SendResponse(Channel& channel,
-                                const Packet& request,
-                                const void* response_struct,
-                                Status status) const {
-  Channel::OutputBuffer response_buffer = channel.AcquireBuffer();
-  std::span payload_buffer = response_buffer.payload(request);
-
-  StatusWithSize encoded =
-      serde_.EncodeResponse(response_struct, payload_buffer);
-
-  if (encoded.ok()) {
-    Packet response = Packet::Response(request);
-
-    response.set_payload(payload_buffer.first(encoded.size()));
-    response.set_status(status);
-    pw::Status send_status = channel.Send(response_buffer, response);
-    if (send_status.ok()) {
-      return;
-    }
-
-    PW_LOG_WARN("Failed to send response packet for channel %u, status %u",
-                unsigned(channel.id()),
-                send_status.code());
-
-    // Re-acquire the buffer to encode an error packet.
-    response_buffer = channel.AcquireBuffer();
-  } else {
-    PW_LOG_WARN(
-        "Nanopb failed to encode response packet for channel %u, status %u",
-        unsigned(channel.id()),
-        encoded.status().code());
-  }
-  channel.Send(response_buffer,
-               Packet::ServerError(request, Status::Internal()));
-}
-
 }  // namespace internal
 }  // namespace pw::rpc
diff --git a/pw_rpc/nanopb/method_lookup_test.cc b/pw_rpc/nanopb/method_lookup_test.cc
index 6017ff6..d54e2fe 100644
--- a/pw_rpc/nanopb/method_lookup_test.cc
+++ b/pw_rpc/nanopb/method_lookup_test.cc
@@ -26,6 +26,12 @@
     return StatusWithSize(123);
   }
 
+  void TestAnotherUnaryRpc(ServerContext&,
+                           const pw_rpc_test_TestRequest&,
+                           NanopbServerResponder<pw_rpc_test_TestResponse>&) {
+    called_async_unary_method = true;
+  }
+
   void TestServerStreamRpc(ServerContext&,
                            const pw_rpc_test_TestRequest&,
                            ServerWriter<pw_rpc_test_TestStreamResponse>&) {
@@ -43,6 +49,7 @@
     called_bidirectional_streaming_method = true;
   }
 
+  bool called_async_unary_method = false;
   bool called_server_streaming_method = false;
   bool called_client_streaming_method = false;
   bool called_bidirectional_streaming_method = false;
@@ -56,6 +63,10 @@
     return Status::Unauthenticated();
   }
 
+  void TestAnotherUnaryRpc(ServerContext&, ConstByteSpan, RawServerResponder&) {
+    called_async_unary_method = true;
+  }
+
   void TestServerStreamRpc(ServerContext&, ConstByteSpan, RawServerWriter&) {
     called_server_streaming_method = true;
   }
@@ -70,18 +81,26 @@
     called_bidirectional_streaming_method = true;
   }
 
+  bool called_async_unary_method = false;
   bool called_server_streaming_method = false;
   bool called_client_streaming_method = false;
   bool called_bidirectional_streaming_method = false;
 };
 
-TEST(MixedService1, CallRawMethod_Unary) {
+TEST(MixedService1, CallRawMethod_SyncUnary) {
   PW_RAW_TEST_METHOD_CONTEXT(MixedService1, TestUnaryRpc) context;
   StatusWithSize sws = context.call({});
   EXPECT_TRUE(sws.ok());
   EXPECT_EQ(123u, sws.size());
 }
 
+TEST(MixedService1, CallNanopbMethod_AsyncUnary) {
+  PW_NANOPB_TEST_METHOD_CONTEXT(MixedService1, TestAnotherUnaryRpc) context;
+  ASSERT_FALSE(context.service().called_async_unary_method);
+  context.call({});
+  EXPECT_TRUE(context.service().called_async_unary_method);
+}
+
 TEST(MixedService1, CallNanopbMethod_ServerStreaming) {
   PW_NANOPB_TEST_METHOD_CONTEXT(MixedService1, TestServerStreamRpc) context;
   ASSERT_FALSE(context.service().called_server_streaming_method);
@@ -104,12 +123,19 @@
   EXPECT_TRUE(context.service().called_bidirectional_streaming_method);
 }
 
-TEST(MixedService2, CallNanopbMethod_Unary) {
+TEST(MixedService2, CallNanopbMethod_SyncUnary) {
   PW_NANOPB_TEST_METHOD_CONTEXT(MixedService2, TestUnaryRpc) context;
   Status status = context.call({});
   EXPECT_EQ(Status::Unauthenticated(), status);
 }
 
+TEST(MixedService2, CallRawMethod_AsyncUnary) {
+  PW_RAW_TEST_METHOD_CONTEXT(MixedService2, TestAnotherUnaryRpc) context;
+  ASSERT_FALSE(context.service().called_async_unary_method);
+  context.call({});
+  EXPECT_TRUE(context.service().called_async_unary_method);
+}
+
 TEST(MixedService2, CallRawMethod_ServerStreaming) {
   PW_RAW_TEST_METHOD_CONTEXT(MixedService2, TestServerStreamRpc) context;
   ASSERT_FALSE(context.service().called_server_streaming_method);
diff --git a/pw_rpc/nanopb/method_test.cc b/pw_rpc/nanopb/method_test.cc
index 52638c4..c00370c 100644
--- a/pw_rpc/nanopb/method_test.cc
+++ b/pw_rpc/nanopb/method_test.cc
@@ -43,6 +43,14 @@
     return Status();
   }
 
+  void AsyncUnary(ServerContext&,
+                  const FakePb&,
+                  NanopbServerResponder<FakePb>&) {}
+
+  static void StaticAsyncUnary(ServerContext&,
+                               const FakePb&,
+                               NanopbServerResponder<FakePb>&) {}
+
   Status UnaryWrongArg(ServerContext&, FakePb&, FakePb&) { return Status(); }
 
   static void StaticUnaryVoidReturn(ServerContext&, const FakePb&, FakePb&) {}
@@ -127,12 +135,12 @@
 NanopbServerReaderWriter<pw_rpc_test_TestRequest, pw_rpc_test_TestResponse>
     last_reader_writer;
 
-Status AddFive(ServerContext&,
-               const pw_rpc_test_TestRequest& request,
-               pw_rpc_test_TestResponse& response) {
+void AddFive(ServerContext&,
+             const pw_rpc_test_TestRequest& request,
+             NanopbServerResponder<pw_rpc_test_TestResponse>& responder) {
   last_request = request;
-  response.value = request.integer + 5;
-  return Status::Unauthenticated();
+  responder.Finish({.value = static_cast<int32_t>(request.integer + 5)},
+                   Status::Unauthenticated());
 }
 
 Status DoNothing(ServerContext&, const pw_rpc_test_Empty&, pw_rpc_test_Empty&) {
@@ -164,9 +172,9 @@
   FakeService(uint32_t id) : Service(id, kMethods) {}
 
   static constexpr std::array<NanopbMethodUnion, 5> kMethods = {
-      NanopbMethod::Unary<DoNothing>(
+      NanopbMethod::SynchronousUnary<DoNothing>(
           10u, pw_rpc_test_Empty_fields, pw_rpc_test_Empty_fields),
-      NanopbMethod::Unary<AddFive>(
+      NanopbMethod::AsynchronousUnary<AddFive>(
           11u, pw_rpc_test_TestRequest_fields, pw_rpc_test_TestResponse_fields),
       NanopbMethod::ServerStreaming<StartStream>(
           12u, pw_rpc_test_TestRequest_fields, pw_rpc_test_TestResponse_fields),
diff --git a/pw_rpc/nanopb/public/pw_rpc/nanopb/internal/method.h b/pw_rpc/nanopb/public/pw_rpc/nanopb/internal/method.h
index 24cd2dc..9c99a23 100644
--- a/pw_rpc/nanopb/public/pw_rpc/nanopb/internal/method.h
+++ b/pw_rpc/nanopb/public/pw_rpc/nanopb/internal/method.h
@@ -42,6 +42,11 @@
                                       Response&);
 
 template <typename Request, typename Response>
+using NanopbAsynchronousUnary = void(ServerContext&,
+                                     const Request&,
+                                     NanopbServerResponder<Response>&);
+
+template <typename Request, typename Response>
 using NanopbServerStreaming = void(ServerContext&,
                                    const Request&,
                                    NanopbServerWriter<Response>&);
@@ -54,7 +59,7 @@
 using NanopbBidirectionalStreaming =
     void(ServerContext&, NanopbServerReaderWriter<Request, Response>&);
 
-// MethodTraits specialization for a static unary method.
+// MethodTraits specialization for a static synchronous unary method.
 template <typename Req, typename Resp>
 struct MethodTraits<NanopbSynchronousUnary<Req, Resp>*> {
   using Implementation = NanopbMethod;
@@ -62,17 +67,33 @@
   using Response = Resp;
 
   static constexpr MethodType kType = MethodType::kUnary;
+  static constexpr bool kSynchronous = true;
+
   static constexpr bool kServerStreaming = false;
   static constexpr bool kClientStreaming = false;
 };
 
-// MethodTraits specialization for a unary method.
+// MethodTraits specialization for a synchronous unary method.
 template <typename T, typename Req, typename Resp>
 struct MethodTraits<NanopbSynchronousUnary<Req, Resp>(T::*)>
     : MethodTraits<NanopbSynchronousUnary<Req, Resp>*> {
   using Service = T;
 };
 
+// MethodTraits specialization for a static asynchronous unary method.
+template <typename Req, typename Resp>
+struct MethodTraits<NanopbAsynchronousUnary<Req, Resp>*>
+    : MethodTraits<NanopbSynchronousUnary<Req, Resp>*> {
+  static constexpr bool kSynchronous = false;
+};
+
+// MethodTraits specialization for an asynchronous unary method.
+template <typename T, typename Req, typename Resp>
+struct MethodTraits<NanopbAsynchronousUnary<Req, Resp>(T::*)>
+    : MethodTraits<NanopbSynchronousUnary<Req, Resp>(T::*)> {
+  static constexpr bool kSynchronous = false;
+};
+
 // MethodTraits specialization for a static server streaming method.
 template <typename Req, typename Resp>
 struct MethodTraits<NanopbServerStreaming<Req, Resp>*> {
@@ -151,9 +172,10 @@
 
   // Creates a NanopbMethod for a synchronous unary RPC.
   template <auto kMethod>
-  static constexpr NanopbMethod Unary(uint32_t id,
-                                      NanopbMessageDescriptor request,
-                                      NanopbMessageDescriptor response) {
+  static constexpr NanopbMethod SynchronousUnary(
+      uint32_t id,
+      NanopbMessageDescriptor request,
+      NanopbMessageDescriptor response) {
     // Define a wrapper around the user-defined function that takes the
     // request and response protobuf structs as void*. This wrapper is stored
     // generically in the Function union, defined below.
@@ -176,6 +198,33 @@
         response);
   }
 
+  // Creates a NanopbMethod for an asynchronous unary RPC.
+  template <auto kMethod>
+  static constexpr NanopbMethod AsynchronousUnary(
+      uint32_t id,
+      NanopbMessageDescriptor request,
+      NanopbMessageDescriptor response) {
+    // Define a wrapper around the user-defined function that takes the
+    // request and response protobuf structs as void*. This wrapper is stored
+    // generically in the Function union, defined below.
+    //
+    // In optimized builds, the compiler inlines the user-defined function into
+    // this wrapper, elminating any overhead.
+    constexpr UnaryRequestFunction wrapper =
+        [](CallContext& call, const void* req, GenericNanopbResponder& resp) {
+          return CallMethodImplFunction<kMethod>(
+              call,
+              *static_cast<const Request<kMethod>*>(req),
+              static_cast<NanopbServerResponder<Response<kMethod>>&>(resp));
+        };
+    return NanopbMethod(
+        id,
+        AsynchronousUnaryInvoker<AllocateSpaceFor<Request<kMethod>>()>,
+        Function{.unary_request = wrapper},
+        request,
+        response);
+  }
+
   // Creates a NanopbMethod for a server-streaming RPC.
   template <auto kMethod>
   static constexpr NanopbMethod ServerStreaming(
@@ -319,6 +368,21 @@
         call, request, &request_struct, &response_struct);
   }
 
+  // Invoker function for asynchronous unary RPCs. Allocates space for a request
+  // struct. Ignores the payload buffer since resposnes are sent through the
+  // NanopbServerResponder.
+  template <size_t kRequestSize>
+  static void AsynchronousUnaryInvoker(const Method& method,
+                                       CallContext& call,
+                                       const Packet& request) {
+    _PW_RPC_NANOPB_STRUCT_STORAGE_CLASS
+    std::aligned_storage_t<kRequestSize, alignof(std::max_align_t)>
+        request_struct{};
+
+    static_cast<const NanopbMethod&>(method).CallUnaryRequest(
+        call, MethodType::kUnary, request, &request_struct);
+  }
+
   // Invoker function for server streaming RPCs. Allocates space for a request
   // struct. Ignores the payload buffer since resposnes are sent through the
   // NanopbServerWriter.
@@ -361,12 +425,6 @@
                      const Packet& request,
                      void* proto_struct) const;
 
-  // Encodes a response and sends it over the provided channel.
-  void SendResponse(Channel& channel,
-                    const Packet& request,
-                    const void* response_struct,
-                    Status status) const;
-
   // Stores the user-defined RPC in a generic wrapper.
   Function function_;
 
diff --git a/pw_rpc/nanopb/public/pw_rpc/nanopb/server_reader_writer.h b/pw_rpc/nanopb/public/pw_rpc/nanopb/server_reader_writer.h
index 303825c..2cc36bf 100644
--- a/pw_rpc/nanopb/public/pw_rpc/nanopb/server_reader_writer.h
+++ b/pw_rpc/nanopb/public/pw_rpc/nanopb/server_reader_writer.h
@@ -45,19 +45,19 @@
   GenericNanopbResponder(const CallContext& call, MethodType type)
       : internal::Call(call, type) {}
 
+  Status SendResponse(const void* response, Status status) {
+    return SendClientStreamOrResponse(response, &status);
+  }
+
  protected:
   Status SendClientStream(const void* response) {
     return SendClientStreamOrResponse(response, nullptr);
   }
 
-  Status SendResponse(const void* response, Status status) {
-    return SendClientStreamOrResponse(response, &status);
-  }
-
   void DecodeRequest(ConstByteSpan payload, void* request_struct) const;
 
  private:
-  Status SendClientStreamOrResponse(const void* response, Status* status);
+  Status SendClientStreamOrResponse(const void* response, const Status* status);
 };
 
 // The BaseNanopbServerReader serves as the base for the ServerReader and
@@ -276,6 +276,58 @@
       : internal::GenericNanopbResponder(call, MethodType::kServerStreaming) {}
 };
 
+template <typename Response>
+class NanopbServerResponder : private internal::GenericNanopbResponder {
+ public:
+  // Creates a NanopbServerResponder that is ready to send a response for a
+  // particular RPC. This can be used for testing or to send responses to an RPC
+  // that has not been started by a client.
+  template <auto kMethod, uint32_t kMethodId, typename ServiceImpl>
+  [[nodiscard]] static NanopbServerResponder Open(Server& server,
+                                                  uint32_t channel_id,
+                                                  ServiceImpl& service) {
+    static_assert(
+        std::is_same_v<Response, internal::Response<kMethod>>,
+        "The response type of a NanopbServerResponder must match the method.");
+    return {internal::OpenCall<kMethod, MethodType::kUnary>(
+        server,
+        channel_id,
+        service,
+        internal::MethodLookup::GetNanopbMethod<ServiceImpl, kMethodId>())};
+  }
+
+  // Allow default construction so that users can declare a variable into which
+  // to move ServerWriters from RPC calls.
+  constexpr NanopbServerResponder()
+      : internal::GenericNanopbResponder(MethodType::kUnary) {}
+
+  NanopbServerResponder(NanopbServerResponder&&) = default;
+  NanopbServerResponder& operator=(NanopbServerResponder&&) = default;
+
+  using internal::GenericNanopbResponder::active;
+  using internal::GenericNanopbResponder::channel_id;
+
+  // Sends the response. Returns the following Status codes:
+  //
+  //   OK - the response was successfully sent
+  //   FAILED_PRECONDITION - the writer is closed
+  //   INTERNAL - pw_rpc was unable to encode the Nanopb protobuf
+  //   other errors - the ChannelOutput failed to send the packet; the error
+  //       codes are determined by the ChannelOutput implementation
+  //
+  Status Finish(const Response& response, Status status = OkStatus()) {
+    return internal::GenericNanopbResponder::SendResponse(&response, status);
+  }
+
+ private:
+  friend class internal::NanopbMethod;
+
+  template <typename, typename, uint32_t>
+  friend class internal::test::InvocationContext;
+
+  NanopbServerResponder(const internal::CallContext& call)
+      : internal::GenericNanopbResponder(call, MethodType::kUnary) {}
+};
 // TODO(hepler): "pw::rpc::ServerWriter" should not be specific to Nanopb.
 template <typename T>
 using ServerWriter = NanopbServerWriter<T>;
diff --git a/pw_rpc/nanopb/public/pw_rpc/nanopb/test_method_context.h b/pw_rpc/nanopb/public/pw_rpc/nanopb/test_method_context.h
index c530f17..77874cd 100644
--- a/pw_rpc/nanopb/public/pw_rpc/nanopb/test_method_context.h
+++ b/pw_rpc/nanopb/public/pw_rpc/nanopb/test_method_context.h
@@ -158,11 +158,16 @@
   UnaryContext(Args&&... args) : Base(std::forward<Args>(args)...) {}
 
   // Invokes the RPC with the provided request. Returns the status.
-  Status call(const Request& request) {
-    Base::output().clear();
-    Response& response = Base::output().AllocateResponse();
-    return CallMethodImplFunction<kMethod>(
-        Base::call_context(), request, response);
+  auto call(const Request& request) {
+    if constexpr (MethodTraits<decltype(kMethod)>::kSynchronous) {
+      Base::output().clear();
+
+      Response& response = Base::output().AllocateResponse();
+      return CallMethodImplFunction<kMethod>(
+          Base::call_context(), request, response);
+    } else {
+      Base::template call<kMethod, NanopbServerResponder<Response>>(request);
+    }
   }
 };
 
diff --git a/pw_rpc/nanopb/server_reader_writer.cc b/pw_rpc/nanopb/server_reader_writer.cc
index 62ab476..c533e73 100644
--- a/pw_rpc/nanopb/server_reader_writer.cc
+++ b/pw_rpc/nanopb/server_reader_writer.cc
@@ -18,31 +18,30 @@
 
 namespace pw::rpc::internal {
 
-Status GenericNanopbResponder::SendClientStreamOrResponse(const void* response,
-                                                          Status* status) {
+Status GenericNanopbResponder::SendClientStreamOrResponse(
+    const void* response, const Status* status) {
   if (!active()) {
     return Status::FailedPrecondition();
   }
 
-  std::span<std::byte> buffer = AcquirePayloadBuffer();
+  std::span<std::byte> payload_buffer = AcquirePayloadBuffer();
 
   // Cast the method to a NanopbMethod. Access the Nanopb
   // serializer/deserializer object and encode the response with it.
-  auto result = static_cast<const internal::NanopbMethod&>(method())
-                    .serde()
-                    .EncodeResponse(response, buffer);
-  if (!result.ok()) {
-    ReleasePayloadBuffer();
+  StatusWithSize result = static_cast<const internal::NanopbMethod&>(method())
+                              .serde()
+                              .EncodeResponse(response, payload_buffer);
 
-    // If the Nanopb encode failed, the channel output may not have provided a
-    // large enough buffer or something went wrong in Nanopb. Return INTERNAL to
-    // indicate that the problem is internal to the server.
-    return Status::Internal();
+  if (!result.ok()) {
+    return CloseAndSendServerError(Status::Internal());
   }
+
+  payload_buffer = payload_buffer.first(result.size());
+
   if (status != nullptr) {
-    return CloseAndSendResponse(buffer.first(result.size()), *status);
+    return CloseAndSendResponse(payload_buffer, *status);
   }
-  return SendPayloadBufferClientStream(buffer.first(result.size()));
+  return SendPayloadBufferClientStream(payload_buffer);
 }
 
 void GenericNanopbResponder::DecodeRequest(ConstByteSpan payload,
diff --git a/pw_rpc/nanopb/server_reader_writer_test.cc b/pw_rpc/nanopb/server_reader_writer_test.cc
index ff8066b..ad624d9 100644
--- a/pw_rpc/nanopb/server_reader_writer_test.cc
+++ b/pw_rpc/nanopb/server_reader_writer_test.cc
@@ -29,6 +29,10 @@
     return OkStatus();
   }
 
+  void TestAnotherUnaryRpc(ServerContext&,
+                           const pw_rpc_test_TestRequest&,
+                           NanopbServerResponder<pw_rpc_test_TestResponse>&) {}
+
   void TestServerStreamRpc(
       ServerContext&,
       const pw_rpc_test_TestRequest&,
diff --git a/pw_rpc/public/pw_rpc/internal/call.h b/pw_rpc/public/pw_rpc/internal/call.h
index a9822c8..065b56d 100644
--- a/pw_rpc/public/pw_rpc/internal/call.h
+++ b/pw_rpc/public/pw_rpc/internal/call.h
@@ -23,6 +23,7 @@
 #include "pw_rpc/internal/channel.h"
 #include "pw_rpc/internal/config.h"
 #include "pw_rpc/internal/method.h"
+#include "pw_rpc/internal/packet.h"
 #include "pw_rpc/method_type.h"
 #include "pw_rpc/service.h"
 #include "pw_status/status.h"
@@ -70,12 +71,18 @@
   // status from sending the packet, or FAILED_PRECONDITION if the Call is not
   // active.
   Status CloseAndSendResponse(std::span<const std::byte> response,
-                              Status status);
+                              Status status) {
+    return CloseAndSendFinalPacket(PacketType::RESPONSE, response, status);
+  }
 
   Status CloseAndSendResponse(Status status) {
     return CloseAndSendResponse({}, status);
   }
 
+  Status CloseAndSendServerError(Status error) {
+    return CloseAndSendFinalPacket(PacketType::SERVER_ERROR, {}, error);
+  }
+
   void HandleError(Status status) {
     Close();
     if (on_error_) {
@@ -162,6 +169,10 @@
   void ReleasePayloadBuffer();
 
  private:
+  Status CloseAndSendFinalPacket(PacketType type,
+                                 std::span<const std::byte> response,
+                                 Status status);
+
   // Removes the RPC from the server & marks as closed. The responder must be
   // active when this is called.
   void Close();
diff --git a/pw_rpc/public/pw_rpc/internal/method_impl_tester.h b/pw_rpc/public/pw_rpc/internal/method_impl_tester.h
index 1d350cc..4221464 100644
--- a/pw_rpc/public/pw_rpc/internal/method_impl_tester.h
+++ b/pw_rpc/public/pw_rpc/internal/method_impl_tester.h
@@ -36,15 +36,17 @@
 // The TestService class must inherit from Service and provide the following
 // methods with valid signatures for RPCs:
 //
-//   - Unary: a valid unary RPC member function
-//   - StaticUnary: valid unary RPC static member function
-//   - ServerStreaming: valid server streaming RPC member function
-//   - StaticServerStreaming: valid server streaming static RPC member function
-//   - ClientStreaming: valid client streaming RPC member function
-//   - StaticClientStreaming: valid client streaming static RPC member function
-//   - BidirectionalStreaming: valid bidirectional streaming RPC member function
+//   - Unary: synchronous unary RPC member function
+//   - StaticUnary: synchronous unary RPC static member function
+//   - AsyncUnary: asynchronous unary RPC member function
+//   - StaticAsyncUnary: asynchronous unary RPC static member function
+//   - ServerStreaming: server streaming RPC member function
+//   - StaticServerStreaming: server streaming static RPC member function
+//   - ClientStreaming: client streaming RPC member function
+//   - StaticClientStreaming: client streaming static RPC member function
+//   - BidirectionalStreaming: bidirectional streaming RPC member function
 //   - StaticBidirectionalStreaming: bidirectional streaming static RPC
-//     member function
+//         member function
 //
 template <typename MethodImpl, typename TestService>
 class MethodImplTests {
@@ -67,6 +69,11 @@
     static_assert(MethodImpl::template matches<&TestService::StaticUnary,
                                                ExtraTypes...>());
 
+    static_assert(MethodImpl::template matches<&TestService::AsyncUnary,
+                                               ExtraTypes...>());
+    static_assert(MethodImpl::template matches<&TestService::StaticAsyncUnary,
+                                               ExtraTypes...>());
+
     static_assert(MethodImpl::template matches<&TestService::ServerStreaming,
                                                ExtraTypes...>());
     static_assert(
@@ -121,20 +128,36 @@
 
     static_assert(MethodTraits<decltype(&TestService::Unary)>::kType ==
                   MethodType::kUnary);
+    static_assert(MethodTraits<decltype(&TestService::Unary)>::kSynchronous);
     static_assert(MethodTraits<decltype(&TestService::StaticUnary)>::kType ==
                   MethodType::kUnary);
     static_assert(
+        MethodTraits<decltype(&TestService::StaticUnary)>::kSynchronous);
+
+    static_assert(MethodTraits<decltype(&TestService::AsyncUnary)>::kType ==
+                  MethodType::kUnary);
+    static_assert(
+        !MethodTraits<decltype(&TestService::AsyncUnary)>::kSynchronous);
+    static_assert(
+        MethodTraits<decltype(&TestService::StaticAsyncUnary)>::kType ==
+        MethodType::kUnary);
+    static_assert(
+        !MethodTraits<decltype(&TestService::StaticAsyncUnary)>::kSynchronous);
+
+    static_assert(
         MethodTraits<decltype(&TestService::ServerStreaming)>::kType ==
         MethodType::kServerStreaming);
     static_assert(
         MethodTraits<decltype(&TestService::StaticServerStreaming)>::kType ==
         MethodType::kServerStreaming);
+
     static_assert(
         MethodTraits<decltype(&TestService::ClientStreaming)>::kType ==
         MethodType::kClientStreaming);
     static_assert(
         MethodTraits<decltype(&TestService::StaticClientStreaming)>::kType ==
         MethodType::kClientStreaming);
+
     static_assert(
         MethodTraits<decltype(&TestService::BidirectionalStreaming)>::kType ==
         MethodType::kBidirectionalStreaming);
@@ -150,42 +173,54 @@
     constexpr bool Pass() const { return true; }
 
     static constexpr MethodImpl kUnaryMethod =
-        MethodImpl::template Unary<&TestService::Unary>(1, extra_args...);
+        MethodImpl::template SynchronousUnary<&TestService::Unary>(
+            1, extra_args...);
     static_assert(kUnaryMethod.id() == 1);
 
     static constexpr MethodImpl kStaticUnaryMethod =
-        MethodImpl::template Unary<&TestService::StaticUnary>(2, extra_args...);
+        MethodImpl::template SynchronousUnary<&TestService::StaticUnary>(
+            2, extra_args...);
     static_assert(kStaticUnaryMethod.id() == 2);
 
+    static constexpr MethodImpl kAsyncUnaryMethod =
+        MethodImpl::template AsynchronousUnary<&TestService::AsyncUnary>(
+            3, extra_args...);
+    static_assert(kAsyncUnaryMethod.id() == 3);
+
+    static constexpr MethodImpl kStaticAsyncUnaryMethod =
+        MethodImpl::template AsynchronousUnary<&TestService::StaticAsyncUnary>(
+            4, extra_args...);
+    static_assert(kStaticAsyncUnaryMethod.id() == 4);
+
     static constexpr MethodImpl kServerStreamingMethod =
         MethodImpl::template ServerStreaming<&TestService::ServerStreaming>(
-            3, extra_args...);
-    static_assert(kServerStreamingMethod.id() == 3);
+            5, extra_args...);
+    static_assert(kServerStreamingMethod.id() == 5);
 
     static constexpr MethodImpl kStaticServerStreamingMethod =
         MethodImpl::template ServerStreaming<
-            &TestService::StaticServerStreaming>(4, extra_args...);
-    static_assert(kStaticServerStreamingMethod.id() == 4);
+            &TestService::StaticServerStreaming>(6, extra_args...);
+    static_assert(kStaticServerStreamingMethod.id() == 6);
 
     static constexpr MethodImpl kClientStreamingMethod =
         MethodImpl::template ClientStreaming<&TestService::ClientStreaming>(
-            5, extra_args...);
-    static_assert(kClientStreamingMethod.id() == 5);
+            7, extra_args...);
+    static_assert(kClientStreamingMethod.id() == 7);
 
     static constexpr MethodImpl kStaticClientStreamingMethod =
         MethodImpl::template ClientStreaming<
-            &TestService::StaticClientStreaming>(6, extra_args...);
-    static_assert(kStaticClientStreamingMethod.id() == 6);
+            &TestService::StaticClientStreaming>(8, extra_args...);
+    static_assert(kStaticClientStreamingMethod.id() == 8);
 
     static constexpr MethodImpl kBidirectionalStreamingMethod =
         MethodImpl::template BidirectionalStreaming<
-            &TestService::BidirectionalStreaming>(7, extra_args...);
-    static_assert(kBidirectionalStreamingMethod.id() == 7);
+            &TestService::BidirectionalStreaming>(9, extra_args...);
+    static_assert(kBidirectionalStreamingMethod.id() == 9);
 
     static constexpr MethodImpl kStaticBidirectionalStreamingMethod =
         MethodImpl::template BidirectionalStreaming<
-            &TestService::StaticBidirectionalStreaming>(8, extra_args...);
-    static_assert(kStaticBidirectionalStreamingMethod.id() == 8);
+            &TestService::StaticBidirectionalStreaming>(10, extra_args...);
+    static_assert(kStaticBidirectionalStreamingMethod.id() == 10);
 
     // Test that there is an Invalid method creation function.
     static constexpr MethodImpl kInvalidMethod = MethodImpl::Invalid();
diff --git a/pw_rpc/public/pw_rpc/internal/method_union.h b/pw_rpc/public/pw_rpc/internal/method_union.h
index dfe0fcb..0977d1f 100644
--- a/pw_rpc/public/pw_rpc/internal/method_union.h
+++ b/pw_rpc/public/pw_rpc/internal/method_union.h
@@ -113,7 +113,13 @@
   if constexpr (MethodTraits<decltype(kMethod)>::kType != kType) {
     return InvalidMethod<kMethod, kType>(id);
   } else if constexpr (kType == MethodType::kUnary) {
-    return MethodImpl::template Unary<kMethod>(id, std::forward<Args>(args)...);
+    if constexpr (MethodTraits<decltype(kMethod)>::kSynchronous) {
+      return MethodImpl::template SynchronousUnary<kMethod>(
+          id, std::forward<Args>(args)...);
+    } else {
+      return MethodImpl::template AsynchronousUnary<kMethod>(
+          id, std::forward<Args>(args)...);
+    }
   } else if constexpr (kType == MethodType::kServerStreaming) {
     return MethodImpl::template ServerStreaming<kMethod>(
         id, std::forward<Args>(args)...);
diff --git a/pw_rpc/public/pw_rpc/internal/test_method_context.h b/pw_rpc/public/pw_rpc/internal/test_method_context.h
index 61aa735..81260b0 100644
--- a/pw_rpc/public/pw_rpc/internal/test_method_context.h
+++ b/pw_rpc/public/pw_rpc/internal/test_method_context.h
@@ -138,11 +138,11 @@
 
   // Invokes the RPC, optionally with a request argument.
   template <auto kMethod, typename T, typename... RequestArg>
-  auto call(RequestArg&&... request) {
+  void call(RequestArg&&... request) {
     static_assert(sizeof...(request) <= 1);
     output_.clear();
     T responder = GetResponder<T>();
-    return CallMethodImplFunction<kMethod>(
+    CallMethodImplFunction<kMethod>(
         call_context(), std::forward<RequestArg>(request)..., responder);
   }
 
diff --git a/pw_rpc/pw_rpc_test_protos/test.proto b/pw_rpc/pw_rpc_test_protos/test.proto
index d792370..513e70f 100644
--- a/pw_rpc/pw_rpc_test_protos/test.proto
+++ b/pw_rpc/pw_rpc_test_protos/test.proto
@@ -33,6 +33,7 @@
 
 service TestService {
   rpc TestUnaryRpc(TestRequest) returns (TestResponse);
+  rpc TestAnotherUnaryRpc(TestRequest) returns (TestResponse);
   rpc TestServerStreamRpc(TestRequest) returns (stream TestStreamResponse);
   rpc TestClientStreamRpc(stream TestRequest) returns (TestStreamResponse);
   rpc TestBidirectionalStreamRpc(stream TestRequest)
diff --git a/pw_rpc/raw/codegen_test.cc b/pw_rpc/raw/codegen_test.cc
index dcf6ba4..e57c98a 100644
--- a/pw_rpc/raw/codegen_test.cc
+++ b/pw_rpc/raw/codegen_test.cc
@@ -67,6 +67,14 @@
     return StatusWithSize(status, test_response.size());
   }
 
+  static void TestAnotherUnaryRpc(ServerContext& ctx,
+                                  ConstByteSpan request,
+                                  RawServerResponder& responder) {
+    ByteSpan response = responder.PayloadBuffer();
+    StatusWithSize sws = TestUnaryRpc(ctx, request, response);
+    responder.Finish(response.first(sws.size()), sws.status());
+  }
+
   void TestServerStreamRpc(ServerContext&,
                            ConstByteSpan request,
                            RawServerWriter& writer) {
@@ -192,6 +200,26 @@
   }
 }
 
+TEST(RawCodegen, Server_InvokeAsyncUnaryRpc) {
+  PW_RAW_TEST_METHOD_CONTEXT(test::TestService, TestAnotherUnaryRpc) context;
+
+  context.call(EncodeRequest(123, OkStatus()));
+  EXPECT_EQ(OkStatus(), context.status());
+
+  protobuf::Decoder decoder(context.response());
+
+  while (decoder.Next().ok()) {
+    switch (static_cast<test::TestResponse::Fields>(decoder.FieldNumber())) {
+      case test::TestResponse::Fields::VALUE: {
+        int32_t value;
+        EXPECT_EQ(OkStatus(), decoder.ReadInt32(&value));
+        EXPECT_EQ(value, 124);
+        break;
+      }
+    }
+  }
+}
+
 TEST(RawCodegen, Server_InvokeServerStreamingRpc) {
   PW_RAW_TEST_METHOD_CONTEXT(test::TestService, TestServerStreamRpc) context;
 
diff --git a/pw_rpc/raw/method.cc b/pw_rpc/raw/method.cc
index 5ef3bed..c2bf186 100644
--- a/pw_rpc/raw/method.cc
+++ b/pw_rpc/raw/method.cc
@@ -24,34 +24,30 @@
 void RawMethod::SynchronousUnaryInvoker(const Method& method,
                                         CallContext& call,
                                         const Packet& request) {
-  Channel::OutputBuffer response_buffer = call.channel().AcquireBuffer();
-  std::span payload_buffer = response_buffer.payload(request);
+  RawServerResponder responder(call);
+  std::span payload_buffer = responder.AcquirePayloadBuffer();
 
   StatusWithSize sws =
       static_cast<const RawMethod&>(method).function_.synchronous_unary(
           call, request.payload(), payload_buffer);
-  Packet response = Packet::Response(request);
 
-  response.set_payload(payload_buffer.first(sws.size()));
-  response.set_status(sws.status());
-  if (call.channel().Send(response_buffer, response).ok()) {
-    return;
-  }
-
-  PW_LOG_WARN(
-      "Failed to send response packet for channel %u; terminating RPC with "
-      "INTERNAL error",
-      unsigned(call.channel().id()));
-  call.channel()
-      .Send(Packet::ServerError(request, Status::Internal()))
+  responder.Finish(payload_buffer.first(sws.size()), sws.status())
       .IgnoreError();
 }
 
+void RawMethod::AsynchronousUnaryInvoker(const Method& method,
+                                         CallContext& call,
+                                         const Packet& request) {
+  RawServerResponder responder(call);
+  static_cast<const RawMethod&>(method).function_.asynchronous_unary(
+      call, request.payload(), responder);
+}
+
 void RawMethod::ServerStreamingInvoker(const Method& method,
                                        CallContext& call,
                                        const Packet& request) {
   RawServerWriter server_writer(call);
-  static_cast<const RawMethod&>(method).function_.unary_request(
+  static_cast<const RawMethod&>(method).function_.server_streaming(
       call, request.payload(), server_writer);
 }
 
diff --git a/pw_rpc/raw/method_test.cc b/pw_rpc/raw/method_test.cc
index 0f7eaa8..7b37e9d 100644
--- a/pw_rpc/raw/method_test.cc
+++ b/pw_rpc/raw/method_test.cc
@@ -46,6 +46,12 @@
     return StatusWithSize(0);
   }
 
+  void AsyncUnary(ServerContext&, ConstByteSpan, RawServerResponder&) {}
+
+  static void StaticAsyncUnary(ServerContext&,
+                               ConstByteSpan,
+                               RawServerResponder&) {}
+
   StatusWithSize UnaryWrongArg(ServerContext&, ConstByteSpan, ConstByteSpan) {
     return StatusWithSize(0);
   }
@@ -144,7 +150,7 @@
   FakeService(uint32_t id) : Service(id, kMethods) {}
 
   static constexpr std::array<RawMethodUnion, 2> kMethods = {
-      RawMethod::Unary<AddFive>(10u),
+      RawMethod::SynchronousUnary<AddFive>(10u),
       RawMethod::ServerStreaming<StartStream>(11u),
   };
 };
diff --git a/pw_rpc/raw/public/pw_rpc/raw/internal/method.h b/pw_rpc/raw/public/pw_rpc/raw/internal/method.h
index cd58b61..71677c5 100644
--- a/pw_rpc/raw/public/pw_rpc/raw/internal/method.h
+++ b/pw_rpc/raw/public/pw_rpc/raw/internal/method.h
@@ -36,7 +36,7 @@
   }
 
   template <auto kMethod>
-  static constexpr RawMethod Unary(uint32_t id) {
+  static constexpr RawMethod SynchronousUnary(uint32_t id) {
     constexpr SynchronousUnaryFunction wrapper =
         [](CallContext& call, ConstByteSpan req, ByteSpan res) {
           return CallMethodImplFunction<kMethod>(call, req, res);
@@ -46,13 +46,25 @@
   }
 
   template <auto kMethod>
+  static constexpr RawMethod AsynchronousUnary(uint32_t id) {
+    constexpr AsynchronousUnaryFunction wrapper =
+        [](CallContext& call,
+           ConstByteSpan req,
+           RawServerResponder& responder) {
+          return CallMethodImplFunction<kMethod>(call, req, responder);
+        };
+    return RawMethod(
+        id, AsynchronousUnaryInvoker, Function{.asynchronous_unary = wrapper});
+  }
+
+  template <auto kMethod>
   static constexpr RawMethod ServerStreaming(uint32_t id) {
-    constexpr UnaryRequestFunction wrapper =
+    constexpr ServerStreamingFunction wrapper =
         [](CallContext& call, ConstByteSpan request, RawServerWriter& writer) {
           return CallMethodImplFunction<kMethod>(call, request, writer);
         };
     return RawMethod(
-        id, ServerStreamingInvoker, Function{.unary_request = wrapper});
+        id, ServerStreamingInvoker, Function{.server_streaming = wrapper});
   }
 
   template <auto kMethod>
@@ -85,15 +97,20 @@
                                                       ConstByteSpan,
                                                       ByteSpan);
 
-  using UnaryRequestFunction = void (*)(CallContext&,
-                                        ConstByteSpan,
-                                        RawServerWriter&);
+  using AsynchronousUnaryFunction = void (*)(CallContext&,
+                                             ConstByteSpan,
+                                             RawServerResponder&);
+
+  using ServerStreamingFunction = void (*)(CallContext&,
+                                           ConstByteSpan,
+                                           RawServerWriter&);
 
   using StreamRequestFunction = void (*)(CallContext&, RawServerReaderWriter&);
 
   union Function {
     SynchronousUnaryFunction synchronous_unary;
-    UnaryRequestFunction unary_request;
+    AsynchronousUnaryFunction asynchronous_unary;
+    ServerStreamingFunction server_streaming;
     StreamRequestFunction stream_request;
   };
 
@@ -104,6 +121,10 @@
                                       CallContext& call,
                                       const Packet& request);
 
+  static void AsynchronousUnaryInvoker(const Method& method,
+                                       CallContext& call,
+                                       const Packet& request);
+
   static void ServerStreamingInvoker(const Method& method,
                                      CallContext& call,
                                      const Packet& request);
@@ -124,29 +145,48 @@
 using RawSynchronousUnary = StatusWithSize(ServerContext&,
                                            ConstByteSpan,
                                            ByteSpan);
+using RawAsynchronousUnary = void(ServerContext&,
+                                  ConstByteSpan,
+                                  RawServerResponder&);
 using RawServerStreaming = void(ServerContext&,
                                 ConstByteSpan,
                                 RawServerWriter&);
 using RawClientStreaming = void(ServerContext&, RawServerReader&);
 using RawBidirectionalStreaming = void(ServerContext&, RawServerReaderWriter&);
 
-// MethodTraits specialization for a static raw unary method.
+// MethodTraits specialization for a static synchronous raw unary method.
 template <>
 struct MethodTraits<RawSynchronousUnary*> {
   using Implementation = RawMethod;
 
   static constexpr MethodType kType = MethodType::kUnary;
+  static constexpr bool kSynchronous = true;
+
   static constexpr bool kServerStreaming = false;
   static constexpr bool kClientStreaming = false;
 };
 
-// MethodTraits specialization for a raw unary method.
+// MethodTraits specialization for a synchronous raw unary method.
 template <typename T>
 struct MethodTraits<RawSynchronousUnary(T::*)>
     : MethodTraits<RawSynchronousUnary*> {
   using Service = T;
 };
 
+// MethodTraits specialization for a static asynchronous raw unary method.
+template <>
+struct MethodTraits<RawAsynchronousUnary*>
+    : MethodTraits<RawSynchronousUnary*> {
+  static constexpr bool kSynchronous = false;
+};
+
+// MethodTraits specialization for an asynchronous raw unary method.
+template <typename T>
+struct MethodTraits<RawAsynchronousUnary(T::*)>
+    : MethodTraits<RawSynchronousUnary(T::*)> {
+  static constexpr bool kSynchronous = false;
+};
+
 // MethodTraits specialization for a static raw server streaming method.
 template <>
 struct MethodTraits<RawServerStreaming*> {
diff --git a/pw_rpc/raw/public/pw_rpc/raw/server_reader_writer.h b/pw_rpc/raw/public/pw_rpc/raw/server_reader_writer.h
index 0f05454..c973d85 100644
--- a/pw_rpc/raw/public/pw_rpc/raw/server_reader_writer.h
+++ b/pw_rpc/raw/public/pw_rpc/raw/server_reader_writer.h
@@ -100,7 +100,7 @@
   using internal::Call::open;  // Deprecated; renamed to active()
 
  private:
-  friend class internal::RawMethod;
+  friend class internal::RawMethod;  // Needed to construct
 
   template <typename, typename, uint32_t>
   friend class internal::test::InvocationContext;
@@ -188,8 +188,6 @@
   using RawServerReaderWriter::Write;
 
  private:
-  friend class RawServerReaderWriter;  // Needed for conversions.
-
   template <typename, typename, uint32_t>
   friend class internal::test::InvocationContext;
 
@@ -199,4 +197,48 @@
       : RawServerReaderWriter(call, MethodType::kServerStreaming) {}
 };
 
+// The RawServerResponder is used to send a response in a raw unary RPC.
+class RawServerResponder : private RawServerReaderWriter {
+ public:
+  // Creates a RawServerResponder that is ready to send responses for a
+  // particular RPC. This can be used for testing or to send responses to an RPC
+  // that has not been started by a client.
+  template <auto kMethod, uint32_t kMethodId, typename ServiceImpl>
+  [[nodiscard]] static RawServerResponder Open(Server& server,
+                                               uint32_t channel_id,
+                                               ServiceImpl& service) {
+    return {internal::OpenCall<kMethod, MethodType::kUnary>(
+        server,
+        channel_id,
+        service,
+        internal::MethodLookup::GetRawMethod<ServiceImpl, kMethodId>())};
+  }
+
+  constexpr RawServerResponder() : RawServerReaderWriter(MethodType::kUnary) {}
+
+  RawServerResponder(RawServerResponder&&) = default;
+  RawServerResponder& operator=(RawServerResponder&&) = default;
+
+  using RawServerReaderWriter::active;
+  using RawServerReaderWriter::channel_id;
+
+  using RawServerReaderWriter::set_on_error;
+
+  using RawServerReaderWriter::PayloadBuffer;
+  using RawServerReaderWriter::ReleaseBuffer;
+
+  Status Finish(ConstByteSpan response, Status status = OkStatus()) {
+    return CloseAndSendResponse(response, status);
+  }
+
+ private:
+  template <typename, typename, uint32_t>
+  friend class internal::test::InvocationContext;
+
+  friend class internal::RawMethod;
+
+  RawServerResponder(const internal::CallContext& call)
+      : RawServerReaderWriter(call, MethodType::kUnary) {}
+};
+
 }  // namespace pw::rpc
diff --git a/pw_rpc/raw/public/pw_rpc/raw/test_method_context.h b/pw_rpc/raw/public/pw_rpc/raw/test_method_context.h
index e0e098d..0e0825d 100644
--- a/pw_rpc/raw/public/pw_rpc/raw/test_method_context.h
+++ b/pw_rpc/raw/public/pw_rpc/raw/test_method_context.h
@@ -133,13 +133,18 @@
   UnaryContext(Args&&... args) : Base(std::forward<Args>(args)...) {}
 
   // Invokes the RPC with the provided request. Returns RPC's StatusWithSize.
-  StatusWithSize call(ConstByteSpan request) {
-    Base::output().clear();
-    ByteSpan& response = Base::output().AllocateResponse();
-    auto sws = CallMethodImplFunction<kMethod>(
-        Base::call_context(), request, response);
-    response = response.first(sws.size());
-    return sws;
+  auto call(ConstByteSpan request) {
+    if constexpr (MethodTraits<decltype(kMethod)>::kSynchronous) {
+      Base::output().clear();
+
+      ByteSpan& response = Base::output().AllocateResponse();
+      auto sws = CallMethodImplFunction<kMethod>(
+          Base::call_context(), request, response);
+      response = response.first(sws.size());
+      return sws;
+    } else {
+      Base::template call<kMethod, RawServerResponder>(request);
+    }
   }
 };
 
diff --git a/pw_rpc/raw/server_reader_writer_test.cc b/pw_rpc/raw/server_reader_writer_test.cc
index d73fe0c..ab12a6f 100644
--- a/pw_rpc/raw/server_reader_writer_test.cc
+++ b/pw_rpc/raw/server_reader_writer_test.cc
@@ -27,6 +27,9 @@
     return StatusWithSize(0);
   }
 
+  void TestAnotherUnaryRpc(ServerContext&, ConstByteSpan, RawServerResponder&) {
+  }
+
   void TestServerStreamRpc(ServerContext&, ConstByteSpan, RawServerWriter&) {}
 
   void TestClientStreamRpc(ServerContext&, RawServerReader&) {}