• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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"""Functions for working with pw_rpc packets."""
15
16from typing import Optional
17
18from google.protobuf import message
19from pw_status import Status
20
21from pw_rpc.internal import packet_pb2
22
23
24def decode(data: bytes) -> packet_pb2.RpcPacket:
25    packet = packet_pb2.RpcPacket()
26    packet.MergeFromString(data)
27    return packet
28
29
30def decode_payload(packet, payload_type):
31    payload = payload_type()
32    payload.MergeFromString(packet.payload)
33    return payload
34
35
36def _ids(rpc: tuple) -> tuple:
37    return tuple(item if isinstance(item, int) else item.id for item in rpc)
38
39
40def encode_request(rpc: tuple, request: Optional[message.Message]) -> bytes:
41    channel, service, method = _ids(rpc)
42    payload = request.SerializeToString() if request is not None else bytes()
43
44    return packet_pb2.RpcPacket(type=packet_pb2.PacketType.REQUEST,
45                                channel_id=channel,
46                                service_id=service,
47                                method_id=method,
48                                payload=payload).SerializeToString()
49
50
51def encode_response(rpc: tuple, response: message.Message) -> bytes:
52    channel, service, method = _ids(rpc)
53
54    return packet_pb2.RpcPacket(
55        type=packet_pb2.PacketType.RESPONSE,
56        channel_id=channel,
57        service_id=service,
58        method_id=method,
59        payload=response.SerializeToString()).SerializeToString()
60
61
62def encode_client_stream(rpc: tuple, request: message.Message) -> bytes:
63    channel, service, method = _ids(rpc)
64
65    return packet_pb2.RpcPacket(
66        type=packet_pb2.PacketType.CLIENT_STREAM,
67        channel_id=channel,
68        service_id=service,
69        method_id=method,
70        payload=request.SerializeToString()).SerializeToString()
71
72
73def encode_client_error(packet: packet_pb2.RpcPacket, status: Status) -> bytes:
74    return packet_pb2.RpcPacket(type=packet_pb2.PacketType.CLIENT_ERROR,
75                                channel_id=packet.channel_id,
76                                service_id=packet.service_id,
77                                method_id=packet.method_id,
78                                status=status.value).SerializeToString()
79
80
81def encode_cancel(rpc: tuple) -> bytes:
82    channel, service, method = _ids(rpc)
83    return packet_pb2.RpcPacket(type=packet_pb2.PacketType.CLIENT_ERROR,
84                                status=Status.CANCELLED.value,
85                                channel_id=channel,
86                                service_id=service,
87                                method_id=method).SerializeToString()
88
89
90def encode_client_stream_end(rpc: tuple) -> bytes:
91    channel, service, method = _ids(rpc)
92
93    return packet_pb2.RpcPacket(type=packet_pb2.PacketType.CLIENT_STREAM_END,
94                                channel_id=channel,
95                                service_id=service,
96                                method_id=method).SerializeToString()
97
98
99def for_server(packet: packet_pb2.RpcPacket) -> bool:
100    return packet.type % 2 == 0
101