hat.drivers.udp

Asyncio UDP endpoint wrapper

 1"""Asyncio UDP endpoint wrapper"""
 2
 3import functools
 4import warnings
 5import typing
 6
 7from hat.drivers import net
 8
 9
10warnings.warn("replaced with hat.drivers.net", DeprecationWarning,
11              stacklevel=2)
12
13
14Address: typing.TypeAlias = net.UdpAddress
15EndpointInfo: typing.TypeAlias = net.UdpEndpointInfo
16Endpoint: typing.TypeAlias = net.Endpoint
17
18create = functools.partial(net.create_endpoint, net.DatagramType.UDP)
Address: TypeAlias = hat.drivers.net.UdpAddress
EndpointInfo: TypeAlias = hat.drivers.net.UdpEndpointInfo
class Endpoint(hat.aio.group.Resource): 
127class Endpoint(aio.Resource):
128    """UDP or Unix Domain Socket endpoint"""
129
130    @property
131    def async_group(self) -> aio.Group:
132        """Async group"""
133        return self._async_group
134
135    @property
136    def info(self) -> EndpointInfo:
137        """Endpoint info"""
138        return self._protocol.info
139
140    @property
141    def empty(self) -> bool:
142        """Is receive queue empty"""
143        return self._protocol.empty
144
145    def send(self,
146             data: util.Bytes,
147             remote_addr: DatagramAddress | None = None):
148        """Send datagram
149
150        If `remote_addr` is not set, `remote_addr` passed to `create` is used.
151
152        """
153        if not self.is_open:
154            raise ConnectionError()
155
156        self._protocol.send(data, remote_addr)
157
158    async def receive(self) -> tuple[util.Bytes, DatagramAddress | None]:
159        """Receive datagram"""
160        return await self._protocol.receive()

UDP or Unix Domain Socket endpoint

async_group: hat.aio.group.Group 
130    @property
131    def async_group(self) -> aio.Group:
132        """Async group"""
133        return self._async_group

Async group

info: hat.drivers.net.UdpEndpointInfo | hat.drivers.net.UnixEndpointInfo 
135    @property
136    def info(self) -> EndpointInfo:
137        """Endpoint info"""
138        return self._protocol.info

Endpoint info

empty: bool 
140    @property
141    def empty(self) -> bool:
142        """Is receive queue empty"""
143        return self._protocol.empty

Is receive queue empty

def send( self, data: bytes | bytearray | memoryview, remote_addr: hat.drivers.net.UdpAddress | pathlib.Path | None = None): 
145    def send(self,
146             data: util.Bytes,
147             remote_addr: DatagramAddress | None = None):
148        """Send datagram
149
150        If `remote_addr` is not set, `remote_addr` passed to `create` is used.
151
152        """
153        if not self.is_open:
154            raise ConnectionError()
155
156        self._protocol.send(data, remote_addr)

Send datagram

If remote_addr is not set, remote_addr passed to create is used.

async def receive( self) -> tuple[bytes | bytearray | memoryview, hat.drivers.net.UdpAddress | pathlib.Path | None]: 
158    async def receive(self) -> tuple[util.Bytes, DatagramAddress | None]:
159        """Receive datagram"""
160        return await self._protocol.receive()

Receive datagram

async def create( local_addr: hat.drivers.net.UdpAddress | pathlib.Path | None = None, remote_addr: hat.drivers.net.UdpAddress | pathlib.Path | None = None, *, name: str | None = None, receive_queue_size: int = 0, **kwargs) -> 'Endpoint':

Create a new function with partial application of the given arguments and keywords.