constexpr T ConvertToValue(rpc::SynchronousCallResult<T>& result) {
  return std::move(result.response());
}

// RPC service with low-level RPCs for transmitting data. Used for benchmarking
// and testing.
//
// NOTE: the implementation of `BidirectionalEcho` is *not* thread-safe.
class BenchmarkService
    : public pw_rpc::raw::Benchmark::Service<BenchmarkService> {
 public:
  static void UnaryEcho(ConstByteSpan request, RawUnaryResponder& responder);

  void BidirectionalEcho(RawServerReaderWriter& reader_writer);

 private:
  using ReaderWriterId = uint64_t;
  ReaderWriterId AllocateReaderWriterId();

  ReaderWriterId next_reader_writer_id_ = 0;

void BidirectionalEcho(RawServerReaderWriter& reader_writer);

static void UnaryEcho(ConstByteSpan request, RawUnaryResponder& responder);

Status ChangeEncodedChannelId(ByteSpan rpc_packet) {
  static_assert(kNewChannelId < 128u,
                "Channel IDs must be less than 128 to avoid needing to "
                "re-encode the packet");
  return internal::OverwriteChannelId(rpc_packet, kNewChannelId);
}

// Associates an ID with an interface for sending packets.
class Channel : public internal::ChannelBase {
 public:
  // Creates a channel with a static ID. The channel's output can also be
  // static, or it can set to null to allow dynamically opening connections
  // through the channel.
  template <uint32_t kId>
  constexpr static Channel Create(ChannelOutput* output) {
    static_assert(kId != kUnassignedChannelId, "Channel ID cannot be 0");
    return Channel(kId, output);
  }

  // Creates a channel with a static ID from an enum value.

// Creates a dynamically assignable channel without a set ID or output.
constexpr Channel() : internal::ChannelBase(kUnassignedChannelId, nullptr) {}

constexpr static Channel Create(ChannelOutput* output) {
  static_assert(kId != kUnassignedChannelId, "Channel ID cannot be 0");
  return Channel(kId, output);
}

class ChannelOutput {
 public:
  // Returned from MaximumTransmissionUnit() to indicate that this ChannelOutput
  // imposes no limits on the MTU.
  static constexpr size_t kUnlimited = std::numeric_limits<size_t>::max();

  // Creates a channel output with the provided name. The name is used for
  // logging only.
  constexpr ChannelOutput(const char* name) : name_(name) {}

  virtual ~ChannelOutput() = default;

// Creates a channel output with the provided name. The name is used for
// logging only.
constexpr ChannelOutput(const char* name) : name_(name) {}

// Returns the maximum transmission unit that this ChannelOutput supports. If
// the ChannelOutput imposes no limit on the MTU, this function returns
// ChannelOutput::kUnlimited.
virtual size_t MaximumTransmissionUnit() { return kUnlimited; }

// Send() function or any descendent calls. Doing so will result in deadlock!
// RPC APIs may be used by other threads, just not within Send().
//
// The buffer provided in packet must NOT be accessed outside of this
// function. It must be sent immediately or copied elsewhere before the
// function returns.
virtual Status Send(span<const std::byte> buffer)

// Returned from MaximumTransmissionUnit() to indicate that this ChannelOutput
// imposes no limits on the MTU.
static constexpr size_t kUnlimited = std::numeric_limits<size_t>::max();

constexpr const char* name() const { return name_; }

class Client : public internal::Endpoint {
 public:
  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR Client() = default;
#endif  // PW_RPC_DYNAMIC_ALLOCATION

  // Creates a client that uses a set of RPC channels. Channels can be shared
  // between multiple clients and servers.
  _PW_RPC_CONSTEXPR Client(span<Channel> channels) : Endpoint(channels) {}

  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR Client() = default;

//
//   OK - The packet was processed by the client.
//   DATA_LOSS - Failed to decode the packet.
//   INVALID_ARGUMENT - The packet is intended for a server, not a client.
//   UNAVAILABLE - No RPC channel with the requested ID was found.
//
Status ProcessPacket(ConstByteSpan data)

// Class that wraps both an RPC client and a server, simplifying RPC setup when
// a device needs to function as both.
class ClientServer {
 public:
  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR ClientServer() = default;
#endif  // PW_RPC_DYNAMIC_ALLOCATION

  _PW_RPC_CONSTEXPR ClientServer(span<Channel> channels)
      : client_(channels), server_(channels) {}

  // Sends a packet to either the client or the server, depending on its type.

  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR ClientServer() = default;

// Sends a packet to either the client or the server, depending on its type.
Status ProcessPacket(ConstByteSpan packet);

constexpr Client& client() { return client_; }

constexpr Server& server() { return server_; }

/// Extracts the channel ID from a pw_rpc packet.
///
/// @returns @Result{the channel ID}
/// * @DATA_LOSS: The packet is corrupt and the channel ID could not be found.
Result<uint32_t> ExtractChannelId(ConstByteSpan packet);

constexpr MethodId GetMethodId() {
  return internal::WrapMethodId(internal::MethodInfo<kMethod>::kMethodId);
}

constexpr ServiceId GetServiceIdForMethod() {
  return internal::WrapServiceId(internal::MethodInfo<kMethod>::kServiceId);
}

// True for client and bidirectional streaming RPCs.
constexpr bool HasClientStream(MethodType type) {
  return (static_cast<unsigned>(type) &
          static_cast<unsigned>(MethodType::kClientStreaming)) != 0;
}

// True for unary and server streaming RPCs.
constexpr bool HasServerStream(MethodType type) {
  return (static_cast<unsigned>(type) &
          static_cast<unsigned>(MethodType::kServerStreaming)) != 0;
}

/// determine the largest supported payload, even when dynamic allocation is
/// enabled.
///
/// @warning `MaxSafePayloadSize` does NOT account for the channel MTU, which
/// may be smaller. Call `MaxWriteSizeBytes()` on an RPC's call object
/// (reader/writer) to account for channel MTU.
constexpr size_t MaxSafePayloadSize(
    size_t encode_buffer_size = cfg::kEncodingBufferSizeBytes) {
  return encode_buffer_size > internal::Packet::kMinEncodedSizeWithoutPayload
             ? encode_buffer_size -
                   internal::Packet::kMinEncodedSizeWithoutPayload
             : 0;
}

// An identifier for a method.
class MethodId {
 private:
  constexpr explicit MethodId(uint32_t id) : id_(id) {}
  friend constexpr MethodId internal::WrapMethodId(uint32_t id);
  friend constexpr uint32_t internal::UnwrapMethodId(MethodId id);
  uint32_t id_;
};

constexpr const auto& MethodSerde() {
  return internal::MethodInfo<kMethod>::serde();
}

enum class MethodType : unsigned char {
  kUnary = 0b00,
  kServerStreaming = 0b01,
  kClientStreaming = 0b10,
  kBidirectionalStreaming = 0b11,
};

// Metadata about a `pw_rpc` packet.
//
// For now, this metadata structure only includes a limited set of information
// about the contents of a packet, but it may be extended in the future.
class PacketMeta {
 public:
  // Parses the metadata from a serialized packet.
  static Result<PacketMeta> FromBuffer(ConstByteSpan data);
  constexpr uint32_t channel_id() const { return channel_id_; }
  constexpr ServiceId service_id() const { return service_id_; }
  constexpr MethodId method_id() const { return method_id_; }
  constexpr bool destination_is_client() const {
    return destination_ == internal::Packet::kClient;
  }
  constexpr bool destination_is_server() const {
    return destination_ == internal::Packet::kServer;

// Parses the metadata from a serialized packet.
static Result<PacketMeta> FromBuffer(ConstByteSpan data);

constexpr uint32_t channel_id() const { return channel_id_; }

constexpr bool destination_is_client() const {
  return destination_ == internal::Packet::kClient;
}

constexpr bool destination_is_server() const {
  return destination_ == internal::Packet::kServer;
}

constexpr MethodId method_id() const { return method_id_; }

// Note: this `payload` is only valid so long as the original `data` buffer
// passed to `PacketMeta::FromBuffer` remains valid.
constexpr ConstByteSpan payload() const { return payload_; }

constexpr ServiceId service_id() const { return service_id_; }

constexpr bool type_is_client_error() const {
  return type_ == internal::pwpb::PacketType::CLIENT_ERROR;
}

constexpr bool type_is_server_error() const {
  return type_ == internal::pwpb::PacketType::SERVER_ERROR;
}

// Returns the payloads for a particular RPC in a Vector of RPC packets.
//
// Adapts a FilteredView of packets to return payloads instead of packets.
class PayloadsView {
 public:
  class iterator : public containers::WrappedIterator<
                       iterator,
                       internal::test::PacketsView::iterator,
                       ConstByteSpan> {
   public:
    constexpr iterator() = default;

    // Access the payload (rather than packet) with operator* and operator->.
    const ConstByteSpan& operator*() const { return value().payload(); }
    const ConstByteSpan* operator->() const { return &value().payload(); }

std::advance(it, index);

const ConstByteSpan& back() const { return *std::prev(end()); }

auto it = begin();

bool empty() const { return begin() == end(); }

iterator end() const { return iterator(view_.end()); }

// Returns the first/last payload for the RPC. size() must be > 0.
const ConstByteSpan& front() const { return *begin(); }

class iterator : public containers::WrappedIterator<
                     iterator,
                     internal::test::PacketsView::iterator,
                     ConstByteSpan> {
 public:
  constexpr iterator() = default;

  // Access the payload (rather than packet) with operator* and operator->.
  const ConstByteSpan& operator*() const { return value().payload(); }
  const ConstByteSpan* operator->() const { return &value().payload(); }

 private:

constexpr iterator() = default;

// Access the payload (rather than packet) with operator* and operator->.
const ConstByteSpan& operator*() const { return value().payload(); }

// Number of payloads for the specified RPC.
size_t size() const { return view_.size(); }

SynchronousCallResult<Response>::RpcError(Status status) {
  return SynchronousCallResult(internal::SynchronousCallStatus::kRpc, status);
}

class Server : public internal::Endpoint {
 public:
  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR Server() = default;
#endif  // PW_RPC_DYNAMIC_ALLOCATION

  // Creates a client that uses a set of RPC channels. Channels can be shared
  // between multiple clients and servers.
  _PW_RPC_CONSTEXPR Server(span<Channel> channels) : Endpoint(channels) {}

// Returns whether a service is registered.
//
// Calling RegisterService with a registered service will assert. So depending
// on your logic you might want to check if a service is currently registered.
bool IsServiceRegistered(const Service& service) const

// the packet was able to be processed:
//
//   OK - The packet was processed by the server.
//   DATA_LOSS - Failed to decode the packet.
//   INVALID_ARGUMENT - The packet is intended for a client, not a server.
//   UNAVAILABLE - No RPC channel with the requested ID was found.
Status ProcessPacket(ConstByteSpan packet_data)

void RegisterService(Service& service, OtherServices&... services)

  // If dynamic allocation is supported, it is not necessary to preallocate a
  // channels list.
#if PW_RPC_DYNAMIC_ALLOCATION
  _PW_RPC_CONSTEXPR Server() = default;

void UnregisterService(Service& service, OtherServices&... services)

// Base class for all RPC services. This cannot be instantiated directly; use a
// generated subclass instead.
//
// Services store a span of concrete method implementation classes. To support
// different Method implementations, Service stores a base MethodUnion* and the
// size of the concrete MethodUnion object.
class Service : public IntrusiveList<Service>::Item {
 public:
  Service(const Service&) = delete;
  Service(Service&&) = delete;

  Service& operator=(const Service&) = delete;
  Service& operator=(Service&&) = delete;

  ServiceId service_id() const { return internal::WrapServiceId(id_); }

 protected:
  // Note: despite being non-`::internal` and non-`private`, this constructor

Service(const Service&) = delete;

ServiceId service_id() const { return internal::WrapServiceId(id_); }

// An identifier for a service.
//
// Note: this does not identify instances of a service (Servers), only the
// service itself.
class ServiceId {
 private:
  constexpr explicit ServiceId(uint32_t id) : id_(id) {}
  friend constexpr ServiceId internal::WrapServiceId(uint32_t id);
  friend constexpr uint32_t internal::UnwrapServiceId(ServiceId id);
  uint32_t id_;
};

// Class for iterating over RPC statuses associated witha particular RPC. This
// is used to iterate over the user RPC statuses and or protocol errors for a
// particular RPC.
class StatusView {
 public:
  class iterator : public containers::WrappedIterator<
                       iterator,
                       internal::test::PacketsView::iterator,
                       Status> {
   public:
    constexpr iterator() = default;

    // Access the status (rather than packet) with operator* and operator->.
    const Status& operator*() const { return value().status(); }
    const Status* operator->() const { return &value().status(); }

std::advance(it, index);

const Status& back() const { return *std::prev(end()); }

auto it = begin();

bool empty() const { return begin() == end(); }

iterator end() const { return iterator(view_.end()); }

// Returns the first/last payload for the RPC. size() must be > 0.
const Status& front() const { return *begin(); }

class iterator : public containers::WrappedIterator<
                     iterator,
                     internal::test::PacketsView::iterator,
                     Status> {
 public:
  constexpr iterator() = default;

  // Access the status (rather than packet) with operator* and operator->.
  const Status& operator*() const { return value().status(); }
  const Status* operator->() const { return &value().status(); }

 private:

constexpr iterator() = default;

// Access the status (rather than packet) with operator* and operator->.
const Status& operator*() const { return value().status(); }

// Number of statuses in this view.
size_t size() const { return view_.size(); }

SynchronousCallResult<Response> SynchronousCall(
    Client& client,
    uint32_t channel_id,
    const typename internal::MethodInfo<kRpcMethod>::Request& request) {
  return internal::StructSynchronousCall<kRpcMethod, Response>(
      internal::CallFreeFunctionWithCustomResponse<kRpcMethod, Response>(
          client, channel_id, request));
}

SynchronousCallFor(
    Client& client,
    uint32_t channel_id,
    const typename internal::MethodInfo<kRpcMethod>::Request& request,
    chrono::SystemClock::duration timeout) {
  return internal::StructSynchronousCall<kRpcMethod>(
      internal::CallFreeFunction<kRpcMethod>(client, channel_id, request),
      timeout);
}

class SynchronousCallResult {
 public:
  // Error Constructors
  constexpr static SynchronousCallResult Timeout();
  constexpr static SynchronousCallResult RpcError(Status status);

  // Server Response Constructor
  constexpr explicit SynchronousCallResult(Status status, Response response)
      : call_status_(internal::SynchronousCallStatus::kServer),
        status_(status),
        response_(std::move(response)) {}

constexpr static SynchronousCallResult RpcError(Status status);

// Server Response Constructor
constexpr explicit SynchronousCallResult(Status status, Response response)

// Error Constructors
constexpr static SynchronousCallResult Timeout();

// Returns true if there was a timeout, an rpc error or the server returned a
// non-Ok status.
[[nodiscard]] constexpr bool is_error() const;

[[nodiscard]] constexpr bool is_rpc_error() const;

// Returns true if the server responded with a non-Ok status.
[[nodiscard]] constexpr bool is_server_error() const;

[[nodiscard]] constexpr bool is_server_response() const;

[[nodiscard]] constexpr bool is_timeout() const;

// Returns true if the server returned a response with an Ok status.
[[nodiscard]] constexpr bool ok() const;

// SynchronousCallResult<Response>::response()
// SynchronousCallResult<Response>::operator*()
// SynchronousCallResult<Response>::operator->()
//
// Accessors to the held value if `this->is_server_response()`. Otherwise,
// terminates the process.
constexpr const Response& response() const&;

[[nodiscard]] constexpr Status status() const;

SynchronousCallUntil(
    Client& client,
    uint32_t channel_id,
    const typename internal::MethodInfo<kRpcMethod>::Request& request,
    chrono::SystemClock::time_point deadline) {
  return internal::StructSynchronousCall<kRpcMethod>(
      internal::CallFreeFunction<kRpcMethod>(client, channel_id, request),
      deadline);
}

SynchronousCallResult<Response>::Timeout() {
  return SynchronousCallResult(internal::SynchronousCallStatus::kTimeout,
                               Status::DeadlineExceeded());
}

// The Writer class allows writing requests or responses to a streaming RPC.
// ClientWriter, ClientReaderWriter, ServerWriter, and ServerReaderWriter
// classes can be used as a generic Writer.
class Writer {
 public:
  Writer(const Writer&) = delete;
  Writer(Writer&&) = delete;

  Writer& operator=(const Writer&) = delete;
  Writer& operator=(Writer&&) = delete;

  bool active() const;
  uint32_t channel_id() const;

  Status Write(ConstByteSpan payload) PW_LOCKS_EXCLUDED(internal::rpc_lock());

Status Write(ConstByteSpan payload) PW_LOCKS_EXCLUDED(internal::rpc_lock());

Writer(const Writer&) = delete;

bool active() const;

uint32_t channel_id() const;

// server integrations may provide an opportunity to inject a
// ``ChannelManipulator`` for this use case.
//
// A ``ChannelManipulator`` should not set send_packet_, as the consumer of a
// ``ChannelManipulator`` will use ``send_packet`` to insert the provided
// ``ChannelManipulator`` into a packet processing path.
class ChannelManipulator {
 public:
  // The expected function signature of the send callback used by a
  // ChannelManipulator to forward packets to the final destination.
  //
  // The only argument is a byte span containing the RPC packet that should
  // be sent.
  //
  // Returns:
  //   OK - Packet successfully sent.
  //   Other - Failed to send packet.
  using SendCallback = Function<Status(span<const std::byte>)>;

ChannelManipulator()

// Processes an incoming packet before optionally sending it.
//
// Implementations of this method may send the processed packet, multiple
// packets, or no packets at all via the registered `send_packet()`
// handler.
virtual Status ProcessAndSend(span<const std::byte> packet) = 0;

// Sets the true send callback that a ChannelManipulator will use to forward
// packets to the final destination.
//
// This should not be used by a ChannelManipulator. The consumer of a
// ChannelManipulator will set the send callback.
void set_send_packet(SendCallback&& send) { send_packet_ = std::move(send); }

// Initializes logging and the global RPC client for integration testing. Starts
// a background thread that processes incoming.
Status InitializeClient(int argc,
                        char* argv[],
                        const char* usage_args = "PORT");

void SocketClientContext<kMaxTransmissionUnit>::ProcessPackets() {
  constexpr size_t kDecoderBufferSize =
      hdlc::Decoder::RequiredBufferSizeForFrameSize(kMaxTransmissionUnit);
  std::array<std::byte, kDecoderBufferSize> decode_buffer;
  hdlc::Decoder decoder(decode_buffer);

  while (true) {
    std::byte byte[1];
    Result<ByteSpan> read = stream_.Read(byte);

    if (should_terminate_.test()) {
      return;

// Configure options for the socket associated with the client.
int SetClientSockOpt(int level,
                     int optname,
                     const void* optval,
                     unsigned int optlen);

void SetEgressChannelManipulator(ChannelManipulator* new_channel_manipulator);

void SetIngressChannelManipulator(ChannelManipulator* new_channel_manipulator);

class SocketClientContext {
 public:
  constexpr SocketClientContext()
      : rpc_dispatch_thread_handle_(std::nullopt),
        channel_output_(stream_, hdlc::kDefaultRpcAddress, "socket"),
        channel_output_with_manipulator_(channel_output_),
        channel_(
            Channel::Create<kChannelId>(&channel_output_with_manipulator_)),
        client_(span(&channel_, 1)) {}

  Client& client() { return client_; }

class ChannelOutputWithManipulator : public ChannelOutput {
 public:
  ChannelOutputWithManipulator(ChannelOutput& actual_output)
      : ChannelOutput(actual_output.name()),
        actual_output_(actual_output),
        channel_manipulator_(nullptr) {}

  void set_channel_manipulator(ChannelManipulator* new_channel_manipulator) {
    if (new_channel_manipulator != nullptr) {
      new_channel_manipulator->set_send_packet(
          ChannelManipulator::SendCallback(
              [&](ConstByteSpan payload)

ChannelOutputWithManipulator(ChannelOutput& actual_output)

size_t MaximumTransmissionUnit() override {
  return actual_output_.MaximumTransmissionUnit();
}

Status Send(span<const std::byte> buffer) override

void set_channel_manipulator(ChannelManipulator* new_channel_manipulator) {
  if (new_channel_manipulator != nullptr) {
    new_channel_manipulator->set_send_packet(
        ChannelManipulator::SendCallback(
            [&](ConstByteSpan payload)
                __attribute__((no_thread_safety_analysis)) {
                  return actual_output_.Send(payload);
                }));
  }
  channel_manipulator_ = new_channel_manipulator;
}

void SetEgressChannelManipulator(
    ChannelManipulator* new_channel_manipulator) {
  channel_output_with_manipulator_.set_channel_manipulator(
      new_channel_manipulator);
}

void SetIngressChannelManipulator(
    ChannelManipulator* new_channel_manipulator) {
  if (new_channel_manipulator != nullptr) {
    new_channel_manipulator->set_send_packet([&](ConstByteSpan payload) {
      return client_.ProcessPacket(payload);
    });
  }
  ingress_channel_manipulator_ = new_channel_manipulator;
}

// Configure options for the socket associated with the client.
int SetSockOpt(int level,
               int optname,
               const void* optval,
               unsigned int optlen) {
  return stream_.SetSockOpt(level, optname, optval, optlen);
}

constexpr SocketClientContext()

// Connects to the specified host:port and starts a background thread to read
// packets from the socket.
Status Start(const char* host, uint16_t port) {
  PW_TRY(stream_.Connect(host, port));
  rpc_dispatch_thread_handle_.emplace(&SocketClientContext::ProcessPackets,
                                      this);
  return OkStatus();
}

// Terminates the client, joining the RPC dispatch thread.
void Terminate() {
  PW_ASSERT(rpc_dispatch_thread_handle_.has_value());
  should_terminate_.test_and_set();
  // Close the stream to avoid blocking forever on a socket read.
  stream_.Close();
  rpc_dispatch_thread_handle_->join();
}

Client& client() { return client_; }

// Terminates the client, joining the RPC dispatch thread.
void TerminateClient();

// Returns the global RPC client for integration test use.
Client& client();

// Base class for rpc::Channel with internal-only public methods, which are
// hidden in the public derived class.
class ChannelBase {
 public:
  static constexpr uint32_t kUnassignedChannelId = 0;

  // TODO: b/234876441 - Remove the Configure and set_channel_output functions.
  //     Users should call CloseChannel() / OpenChannel() to change a channel.
  //     This ensures calls are properly update and works consistently between
  //     static and dynamic channel allocation.

  // Manually configures a dynamically-assignable channel with a specified ID
  // and output. This is useful when a channel's parameters are not known until
  // runtime. This can only be called once per channel.

constexpr void Close() {
  PW_ASSERT(id_ != kUnassignedChannelId);
  id_ = kUnassignedChannelId;
  output_ = nullptr;
}

constexpr void Configure(uint32_t id, ChannelOutput& output) {
  static_assert(
      !cfg::kDynamicAllocationEnabled<UnusedType>,
      "Configure() may not be used if PW_RPC_DYNAMIC_ALLOCATION is "
      "enabled. Call CloseChannel/OpenChannel on the endpoint instead.");
  PW_ASSERT(id_ == kUnassignedChannelId);
  PW_ASSERT(id != kUnassignedChannelId);
  id_ = id;
  output_ = &output;
}

// Returns the maximum payload size for this channel, factoring in the
// ChannelOutput's MTU and the RPC system's `pw::rpc::MaxSafePayloadSize()`.
size_t MaxWriteSizeBytes() const;

// Invokes ChannelOutput::Send and returns its status. Any non-OK status
// indicates that the Channel is permanently closed.
Status Send(const Packet& packet) PW_EXCLUSIVE_LOCKS_REQUIRED(rpc_lock());

constexpr bool assigned() const { return id_ != kUnassignedChannelId; }

constexpr uint32_t id() const { return id_; }

constexpr void set_channel_output(ChannelOutput& output) {
  static_assert(
      !cfg::kDynamicAllocationEnabled<UnusedType>,
      "set_channel_output() may not be used if PW_RPC_DYNAMIC_ALLOCATION is "
      "enabled. Call CloseChannel/OpenChannel on the endpoint instead.");
  PW_ASSERT(id_ != kUnassignedChannelId);
  output_ = &output;
}

Status OverwriteChannelId(ByteSpan rpc_packet, uint32_t channel_id_under_128);

enum class SynchronousCallStatus {
  kInvalid,
  kTimeout,
  kRpc,
  kServer,
};

constexpr uint32_t UnwrapMethodId(MethodId id);

constexpr uint32_t UnwrapServiceId(ServiceId id);

constexpr MethodId WrapMethodId(uint32_t id);

constexpr ServiceId WrapServiceId(uint32_t id);

// Finds packets of a specified type for a particular method.
class PacketFilter {
 public:
  // Use Channel::kUnassignedChannelId to ignore the channel.
  constexpr PacketFilter(internal::pwpb::PacketType packet_type_1,
                         internal::pwpb::PacketType packet_type_2,
                         uint32_t channel_id,
                         uint32_t service_id,
                         uint32_t method_id)
      : packet_type_1_(packet_type_1),
        packet_type_2_(packet_type_2),
        channel_id_(channel_id),
        service_id_(service_id),

// Use Channel::kUnassignedChannelId to ignore the channel.
constexpr PacketFilter(internal::pwpb::PacketType packet_type_1,
                       internal::pwpb::PacketType packet_type_2,
                       uint32_t channel_id,
                       uint32_t service_id,
                       uint32_t method_id)

constexpr bool operator()(const Packet& packet) const {
  return (packet.type() == packet_type_1_ ||
          packet.type() == packet_type_2_) &&
         (channel_id_ == Channel::kUnassignedChannelId ||
          packet.channel_id() == channel_id_) &&
         packet.service_id() == service_id_ &&
         packet.method_id() == method_id_;
}

constexpr bool SynchronousCallResult<Response>::is_error() const {
  return !ok();
}

constexpr bool SynchronousCallResult<Response>::is_rpc_error() const {
  return call_status_ == internal::SynchronousCallStatus::kRpc;
}

constexpr bool SynchronousCallResult<Response>::is_server_error() const {
  return is_server_response() && !status_.ok();
}

constexpr bool SynchronousCallResult<Response>::is_server_response() const {
  return call_status_ == internal::SynchronousCallStatus::kServer;
}

constexpr bool SynchronousCallResult<Response>::is_timeout() const {
  return call_status_ == internal::SynchronousCallStatus::kTimeout;
}

constexpr bool SynchronousCallResult<Response>::ok() const {
  return is_server_response() && status_.ok();
}

constexpr const Response& SynchronousCallResult<Response>::response() const& {
  PW_ASSERT(is_server_response());
  return response_;
}

constexpr Status SynchronousCallResult<Response>::status() const {
  PW_ASSERT(call_status_ != internal::SynchronousCallStatus::kInvalid);
  return status_;
}

Status SendResponseIfCalled(
    Context& client_context,
    const MethodResponseType<kMethod>& response,
    Status status = OkStatus(),
    chrono::SystemClock::duration timeout =
        chrono::SystemClock::for_at_least(std::chrono::milliseconds(100))) {
  const auto start_time = chrono::SystemClock::now();
  while (chrono::SystemClock::now() - start_time < timeout) {
    const auto count =
        client_context.output().template total_payloads<kMethod>();
    if (count > 0) {
      client_context.server().template SendResponse<kMethod>(response, status);

chrono::SystemClock::for_at_least(std::chrono::milliseconds(100))) {

void WaitForPackets(internal::test::FakeChannelOutput& output,
                    int count,
                    Function&& run_before) {
  sync::CountingSemaphore sem;
  output.set_on_send([&sem](ConstByteSpan, Status) { sem.release(); });

  run_before();

  for (int i = 0; i < count; ++i) {
    PW_ASSERT(sem.try_acquire_for(std::chrono::seconds(kTimeoutSeconds)));
  }