1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
from __future__ import annotations
from typing import List, Optional, Sequence, Tuple
from .. import frames
from ..typing import ExtensionName, ExtensionParameter
__all__ = ["Extension", "ClientExtensionFactory", "ServerExtensionFactory"]
class Extension:
"""
Base class for extensions.
"""
name: ExtensionName
"""Extension identifier."""
def decode(
self,
frame: frames.Frame,
*,
max_size: Optional[int] = None,
) -> frames.Frame:
"""
Decode an incoming frame.
Args:
frame (Frame): incoming frame.
max_size: maximum payload size in bytes.
Returns:
Frame: Decoded frame.
Raises:
PayloadTooBig: if decoding the payload exceeds ``max_size``.
"""
raise NotImplementedError
def encode(self, frame: frames.Frame) -> frames.Frame:
"""
Encode an outgoing frame.
Args:
frame (Frame): outgoing frame.
Returns:
Frame: Encoded frame.
"""
raise NotImplementedError
class ClientExtensionFactory:
"""
Base class for client-side extension factories.
"""
name: ExtensionName
"""Extension identifier."""
def get_request_params(self) -> List[ExtensionParameter]:
"""
Build parameters to send to the server for this extension.
Returns:
List[ExtensionParameter]: Parameters to send to the server.
"""
raise NotImplementedError
def process_response_params(
self,
params: Sequence[ExtensionParameter],
accepted_extensions: Sequence[Extension],
) -> Extension:
"""
Process parameters received from the server.
Args:
params (Sequence[ExtensionParameter]): parameters received from
the server for this extension.
accepted_extensions (Sequence[Extension]): list of previously
accepted extensions.
Returns:
Extension: An extension instance.
Raises:
NegotiationError: if parameters aren't acceptable.
"""
raise NotImplementedError
class ServerExtensionFactory:
"""
Base class for server-side extension factories.
"""
name: ExtensionName
"""Extension identifier."""
def process_request_params(
self,
params: Sequence[ExtensionParameter],
accepted_extensions: Sequence[Extension],
) -> Tuple[List[ExtensionParameter], Extension]:
"""
Process parameters received from the client.
Args:
params (Sequence[ExtensionParameter]): parameters received from
the client for this extension.
accepted_extensions (Sequence[Extension]): list of previously
accepted extensions.
Returns:
Tuple[List[ExtensionParameter], Extension]: To accept the offer,
parameters to send to the client for this extension and an
extension instance.
Raises:
NegotiationError: to reject the offer, if parameters received from
the client aren't acceptable.
"""
raise NotImplementedError
|
