From 3ef820c9e84487450b13f9243ab71cb3a70400ec Mon Sep 17 00:00:00 2001 From: ZipFile Date: Tue, 14 Jul 2026 13:01:40 +0000 Subject: [PATCH] TCP handling logic cleanup --- examples/async_simple_tcp_client.py | 12 +- examples/async_tcp_server.py | 12 +- examples/simple_tcp_client.py | 8 + examples/simple_tcp_server.py | 4 + pythonosc/osc_tcp_server.py | 214 ++++++------------------- pythonosc/parsing/framing.py | 166 +++++++++++++++++++ pythonosc/{ => parsing}/slip.py | 0 pythonosc/tcp_client.py | 165 +++++-------------- pythonosc/test/parsing/test_framing.py | 122 ++++++++++++++ pythonosc/test/test_osc_tcp_server.py | 72 +++++---- pythonosc/test/test_tcp_client.py | 3 +- 11 files changed, 447 insertions(+), 331 deletions(-) create mode 100644 pythonosc/parsing/framing.py rename pythonosc/{ => parsing}/slip.py (100%) create mode 100644 pythonosc/test/parsing/test_framing.py diff --git a/examples/async_simple_tcp_client.py b/examples/async_simple_tcp_client.py index cdadbd3..723861a 100644 --- a/examples/async_simple_tcp_client.py +++ b/examples/async_simple_tcp_client.py @@ -7,8 +7,8 @@ import argparse import asyncio +import logging import random -import sys from pythonosc import tcp_client @@ -47,10 +47,6 @@ async def init_main(): tg.create_task(send_messages(client)) -if sys.version_info >= (3, 7): - asyncio.run(init_main()) -else: - # TODO(python-upgrade): drop this once 3.6 is no longer supported - event_loop = asyncio.get_event_loop() - event_loop.run_until_complete(init_main()) - event_loop.close() +if __name__ == "__main__": + logging.basicConfig(level=logging.DEBUG) + asyncio.run(init_main(), debug=True) diff --git a/examples/async_tcp_server.py b/examples/async_tcp_server.py index 532ff79..71ff273 100644 --- a/examples/async_tcp_server.py +++ b/examples/async_tcp_server.py @@ -1,6 +1,6 @@ import argparse import asyncio -import sys +import logging from pythonosc.dispatcher import Dispatcher from pythonosc.osc_tcp_server import AsyncOSCTCPServer @@ -42,10 +42,6 @@ async def init_main(): tg.create_task(loop()) -if sys.version_info >= (3, 7): - asyncio.run(init_main()) -else: - # TODO(python-upgrade): drop this once 3.6 is no longer supported - event_loop = asyncio.get_event_loop() - event_loop.run_until_complete(init_main()) - event_loop.close() +if __name__ == "__main__": + logging.basicConfig(level=logging.DEBUG) + asyncio.run(init_main(), debug=True) diff --git a/examples/simple_tcp_client.py b/examples/simple_tcp_client.py index 0834742..18c9855 100644 --- a/examples/simple_tcp_client.py +++ b/examples/simple_tcp_client.py @@ -5,11 +5,16 @@ """ import argparse +import logging import random +from random import randint +from zlib import crc32 from pythonosc import tcp_client +from pythonosc.osc_tcp_server import FRAMING_BY_MODE, MODE_1_1 if __name__ == "__main__": + logging.basicConfig(level=logging.DEBUG) parser = argparse.ArgumentParser() parser.add_argument("--ip", default="127.0.0.1", help="The ip of the OSC server") parser.add_argument( @@ -27,6 +32,9 @@ n = random.random() print(f"Sending /filter {n}") client.send_message("/filter", n) + data = bytes(random.randint(0, 255) for i in range(randint(1, 10000))) + print(f"Sending /crc {len(data)} bytes (should be {crc32(data):08X})") + client.send_message("/crc", data) resp = client.get_messages(1) for r in resp: try: diff --git a/examples/simple_tcp_server.py b/examples/simple_tcp_server.py index c1bb0c8..7d549c7 100644 --- a/examples/simple_tcp_server.py +++ b/examples/simple_tcp_server.py @@ -5,7 +5,9 @@ """ import argparse +import logging import math +from zlib import crc32 from pythonosc import osc_tcp_server from pythonosc.dispatcher import Dispatcher @@ -23,6 +25,7 @@ def print_compute_handler(unused_addr, args, volume): if __name__ == "__main__": + logging.basicConfig(level=logging.DEBUG) parser = argparse.ArgumentParser() parser.add_argument("--ip", default="127.0.0.1", help="The ip to listen on") parser.add_argument("--port", type=int, default=5005, help="The port to listen on") @@ -36,6 +39,7 @@ def print_compute_handler(unused_addr, args, volume): dispatcher = Dispatcher() dispatcher.map("/filter", print) + dispatcher.map("/crc", lambda _, v: f"{crc32(v):08X}") dispatcher.map("/volume", print_volume_handler, "Volume") dispatcher.map("/logvolume", print_compute_handler, "Log volume", math.log) diff --git a/pythonosc/osc_tcp_server.py b/pythonosc/osc_tcp_server.py index f3e2778..ca31aa2 100644 --- a/pythonosc/osc_tcp_server.py +++ b/pythonosc/osc_tcp_server.py @@ -33,68 +33,37 @@ # mypy: disable-error-code="attr-defined" import asyncio -import logging import os import socket import socketserver -import struct -from typing import List, Tuple +from typing import Awaitable, Callable, Tuple -from pythonosc import osc_message_builder, slip +from pythonosc import osc_message_builder from pythonosc.dispatcher import Dispatcher +from pythonosc.parsing.framing import ( + BINARY_LENGTH_FRAMING, + SLIP_FRAMING, + Framing, + FramingParser, +) -LOG = logging.getLogger() MODE_1_0 = "1.0" MODE_1_1 = "1.1" +FRAMING_BY_MODE: dict[str, Framing] = { + MODE_1_0: BINARY_LENGTH_FRAMING, + MODE_1_1: SLIP_FRAMING, +} -class _TCPHandler1_0(socketserver.BaseRequestHandler): - """Handles correct OSC1.0 messages. +def get_framing(mode: str) -> Framing: + try: + return FRAMING_BY_MODE[mode] + except KeyError: + raise ValueError(f"Unsupported OSC mode: {mode}") from None - Whether this will be run on its own thread, the server's or a whole new - process depends on the server you instantiated, look at their documentation. - This method is called after a basic sanity check was done on the datagram, - basically whether this datagram looks like an osc message or bundle, - if not the server won't even bother to call it and so no new - threads/processes will be spawned. - """ - - def handle(self) -> None: - LOG.debug("handle OSC 1.0 protocol") - while True: - lengthbuf = self.recvall(4) - if lengthbuf == b"": - break - (length,) = struct.unpack("!I", lengthbuf) - data = self.recvall(length) - if data == b"": - break - - resp = self.server.dispatcher.call_handlers_for_packet( - data, self.client_address - ) - # resp = _call_handlers_for_packet(data, self.server.dispatcher) - for r in resp: - if not isinstance(r, tuple): - r = [r] - msg = osc_message_builder.build_msg(r[0], r[1:]) - b = struct.pack("!I", len(msg.dgram)) - self.request.sendall(b + msg.dgram) - - def recvall(self, count: int) -> bytes: - buf = b"" - while count > 0: - newbuf = self.request.recv(count) - if not newbuf: - return b"" - buf += newbuf - count -= len(newbuf) - return buf - - -class _TCPHandler1_1(socketserver.BaseRequestHandler): - """Handles correct OSC1.1 messages. +class TCPHandler(socketserver.BaseRequestHandler): + """Handles correct OSC messages. Whether this will be run on its own thread, the server's or a whole new process depends on the server you instantiated, look at their documentation. @@ -106,14 +75,12 @@ class _TCPHandler1_1(socketserver.BaseRequestHandler): """ def handle(self) -> None: - LOG.debug("handle OSC 1.1 protocol") - while True: - packets = self.recvall() - if not packets: - break - - for p in packets: - # resp = _call_handlers_for_packet(p, self.server.dispatcher) + assert isinstance(self.server, OSCTCPServer) + + parser = FramingParser(self.server.framing) + + while chunk := self.request.recv(16384): + for p in parser.feed(chunk): resp = self.server.dispatcher.call_handlers_for_packet( p, self.client_address ) @@ -121,27 +88,15 @@ def handle(self) -> None: if not isinstance(r, tuple): r = [r] msg = osc_message_builder.build_msg(r[0], r[1:]) - self.request.sendall(slip.encode(msg.dgram)) - - def recvall(self) -> List[bytes]: - buf = self.request.recv(4096) - if not buf: - return [] - # If the last byte is not an END marker there could be more data coming - while buf[-1] != 192: - newbuf = self.request.recv(4096) - if not newbuf: - # Maybe should raise an exception here? - break - buf += newbuf - - packets = [slip.decode(p) for p in buf.split(slip.END_END)] - return packets + self.request.sendall(self.server.framing.encode(msg.dgram)) class OSCTCPServer(socketserver.TCPServer): """Superclass for different flavors of OSCTCPServer""" + dispatcher: Dispatcher + framing: Framing + def __init__( self, server_address: Tuple[str | bytes | bytearray, int], @@ -150,9 +105,8 @@ def __init__( family: socket.AddressFamily | None = None, ): self.request_queue_size = 300 - self.mode = mode - if mode not in [MODE_1_0, MODE_1_1]: - raise ValueError("OSC Mode must be '1.0' or '1.1'") + self.dispatcher = dispatcher + self.framing = get_framing(mode) if family is not None: self.address_family = family @@ -171,16 +125,7 @@ def __init__( # Fallback to default if resolution fails pass - if self.mode == MODE_1_0: - super().__init__(server_address, _TCPHandler1_0) - else: - super().__init__(server_address, _TCPHandler1_1) - self._dispatcher = dispatcher - - @property - def dispatcher(self): - """Dispatcher accessor for handlers to dispatch osc messages.""" - return self._dispatcher + super().__init__(server_address, TCPHandler) class BlockingOSCTCPServer(OSCTCPServer): @@ -230,8 +175,8 @@ def __init__( """ self._port = port self._server_address = server_address - self._dispatcher = dispatcher - self._mode = mode + self.dispatcher = dispatcher + self.framing = get_framing(mode) async def __aenter__(self): return self @@ -245,9 +190,6 @@ async def start(self) -> None: self.handle, self._server_address, self._port ) - addrs = ", ".join(str(sock.getsockname()) for sock in self._server.sockets) - LOG.debug(f"Serving on {addrs}") - async with self._server: await self._server.serve_forever() @@ -255,86 +197,28 @@ async def stop(self) -> None: self._server.close() await self._server.wait_closed() - @property - def dispatcher(self): - return self._dispatcher - async def handle( self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter ) -> None: + try: + await self._handle(reader, writer) + finally: + if writer.can_write_eof(): + writer.write_eof() + writer.close() + await writer.wait_closed() + + async def _handle( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + parser = FramingParser(self.framing) client_address = ("", 0) sock = writer.transport.get_extra_info("socket") if sock is not None: client_address = sock.getpeername() - if self._mode == MODE_1_1: - await self.handle_1_1(reader, writer, client_address) - else: - await self.handle1_0(reader, writer, client_address) - writer.write_eof() - LOG.debug("Close the connection") - writer.close() - await writer.wait_closed() - - async def handle1_0( - self, - reader: asyncio.StreamReader, - writer: asyncio.StreamWriter, - client_address: Tuple[str, int], - ) -> None: - LOG.debug("Incoming socket open 1.0") - while True: - try: - buf = await reader.read(4) - except Exception as e: - LOG.exception("Read error", e) - return - if buf == b"": - break - (length,) = struct.unpack("!I", buf) - buf = b"" - while length > 0: - newbuf = await reader.read(length) - if not newbuf: - break - buf += newbuf - length -= len(newbuf) - - result = await self.dispatcher.async_call_handlers_for_packet( - buf, client_address - ) - for r in result: - if not isinstance(r, tuple): - r = [r] - msg = osc_message_builder.build_msg(r[0], r[1:]) - b = struct.pack("!I", len(msg.dgram)) - writer.write(b + msg.dgram) - await writer.drain() - - async def handle_1_1( - self, - reader: asyncio.StreamReader, - writer: asyncio.StreamWriter, - client_address: Tuple[str, int], - ) -> None: - LOG.debug("Incoming socket open 1.1") - while True: - try: - buf = await reader.read(4096) - except Exception as e: - LOG.exception("Read error", e) - return - if buf == b"": - break - while len(buf) > 0 and buf[-1] != 192: - newbuf = await reader.read(4096) - if not newbuf: - # Maybe should raise an exception here? - break - buf += newbuf - - packets = [slip.decode(p) for p in buf.split(slip.END_END)] - for p in packets: + while chunk := await reader.read(16384): + for p in parser.feed(chunk): result = await self.dispatcher.async_call_handlers_for_packet( p, client_address ) @@ -342,5 +226,5 @@ async def handle_1_1( if not isinstance(r, tuple): r = [r] msg = osc_message_builder.build_msg(r[0], r[1:]) - writer.write(slip.encode(msg.dgram)) + writer.write(self.framing.encode(msg.dgram)) await writer.drain() diff --git a/pythonosc/parsing/framing.py b/pythonosc/parsing/framing.py new file mode 100644 index 0000000..6e19dee --- /dev/null +++ b/pythonosc/parsing/framing.py @@ -0,0 +1,166 @@ +"""OSC message codecs for different framing strategies.""" + +from abc import ABCMeta, abstractmethod +from struct import pack, unpack_from + +from pythonosc.parsing.slip import ( + END, + END_END, + decode as slip_decode, + encode as slip_encode, +) + + +class Framing(metaclass=ABCMeta): + """Base class for OSC stream framing strategies.""" + + @abstractmethod + def encode(self, dgram: bytes) -> bytes: + """Encode a single OSC datagram for transmission on a byte stream. + + Args: + dgram: Raw OSC datagram bytes. + + Returns: + The framed byte sequence to write to the stream. + """ + + @abstractmethod + def decode_stream( + self, buf: bytes | bytearray, start: int = 0 + ) -> tuple[list[bytes], int]: + """Decode complete OSC datagrams from a byte buffer. + + Args: + buf: Bytes received from a stream. + start: The index in ``buf`` to start decoding from. + + Returns: + A tuple ``(messages, bytes consumed)``. + """ + + +class SLIPFraming(Framing): + """SLIP (RFC 1055) framing for OSC 1.1. + + See the "OSC Delivery Specification 1.1" section from the 2009 NIME paper for details. + """ + + def encode(self, dgram: bytes) -> bytes: + return slip_encode(dgram) + + def decode_stream( + self, buf: bytes | bytearray, start: int = 0 + ) -> tuple[list[bytes], int]: + pos = start + messages = [] + + while (end := buf.find(END_END, pos)) >= 0: + if end > pos: + messages.append(slip_decode(bytes(buf[pos:end]))) + + pos = end + len(END) + + if buf.endswith(END): + messages.append(slip_decode(bytes(buf[pos:end]))) + pos = len(buf) + + return messages, pos - start + + +class BinaryLengthFraming(Framing): + """Length-prefixed framing for OSC 1.0. + + See the "OSC Packets" section from OSC 1.0 Specification for details. + """ + + def encode(self, dgram: bytes) -> bytes: + if dgram: + return pack("!I", len(dgram)) + dgram + return b"" + + def decode_stream( + self, buf: bytes | bytearray, start: int = 0 + ) -> tuple[list[bytes], int]: + messages = [] + pos = start + + while (data_pos := pos + 4) < len(buf): + (length,) = unpack_from("!I", buf, pos) + end = data_pos + length + + if end > len(buf): + break + + messages.append(bytes(buf[data_pos:end])) + + pos = end + + return messages, pos - start + + +class FramingParser: + """Stateful stream decoder for (framed) OSC byte stream. + + Not safe for concurrent use. + + Args: + framing: Strategy used to decode buffered stream data. + max_buffer_size: Maximum size of the internal buffer. + """ + + __slots__ = ("buf", "start", "framing", "max_buffer_size") + + def __init__(self, framing: Framing, max_buffer_size: int = 1048576) -> None: + self.buf = bytearray() + self.start = 0 + self.framing = framing + self.max_buffer_size = max_buffer_size + + def feed(self, chunk: bytes | bytearray | memoryview) -> list[bytes]: + """Append stream data to the internal buffer and return any complete OSC datagrams. + + Args: + chunk: Newly received bytes from the stream. + + Returns: + A list of complete, decoded OSC datagrams. + """ + + new_buf_len = len(self.buf) + len(chunk) + + if (pending := new_buf_len - self.start) > self.max_buffer_size: + raise ValueError(f"Buffer is too large: {pending} > {self.max_buffer_size}") + + if new_buf_len > self.max_buffer_size: + del self.buf[: self.start] + self.start = 0 + + self.buf.extend(chunk) + + messages, consumed = self.framing.decode_stream(self.buf, self.start) + available = len(self.buf) - self.start + + assert 0 <= consumed <= available, ( + f"misbehaving framing decoder: {consumed=} {available=}" + ) + + self.start += consumed + + return messages + + @property + def remaining(self) -> bytes: + """Return any unparsed bytes.""" + + return bytes(self.buf[self.start :]) + + def reset(self) -> None: + """Clear the internal buffer (discard unparsed data).""" + + del self.buf[:] + self.start = 0 + + +SLIP_FRAMING: Framing = SLIPFraming() +BINARY_LENGTH_FRAMING: Framing = BinaryLengthFraming() diff --git a/pythonosc/slip.py b/pythonosc/parsing/slip.py similarity index 100% rename from pythonosc/slip.py rename to pythonosc/parsing/slip.py diff --git a/pythonosc/tcp_client.py b/pythonosc/tcp_client.py index 4d4d858..17a2e4f 100644 --- a/pythonosc/tcp_client.py +++ b/pythonosc/tcp_client.py @@ -2,19 +2,19 @@ import asyncio import socket -import struct -from typing import AsyncGenerator, Generator, Iterable, List, Union +from contextlib import suppress +from typing import Any, AsyncGenerator, Awaitable, Generator, Iterable, List, Union -from pythonosc import slip from pythonosc.dispatcher import Dispatcher from pythonosc.osc_bundle import OscBundle from pythonosc.osc_message import OscMessage from pythonosc.osc_message_builder import ArgValue, build_msg -from pythonosc.osc_tcp_server import MODE_1_1 +from pythonosc.osc_tcp_server import MODE_1_1, get_framing +from pythonosc.parsing.framing import FramingParser -class TCPClient(object): - """Async OSC client to send :class:`OscMessage` or :class:`OscBundle` via TCP""" +class TCPClient: + """OSC client to send :class:`OscMessage` or :class:`OscBundle` via TCP""" def __init__( self, @@ -32,11 +32,13 @@ def __init__( family: address family parameter (passed to socket.getaddrinfo) timeout: Default timeout in seconds for socket operations """ + self.address = address self.port = port self.family = family - self.mode = mode self._timeout = timeout + self._framing = get_framing(mode) + self._parser = FramingParser(self._framing) self.socket = socket.socket(self.family, socket.SOCK_STREAM) self.socket.settimeout(timeout) self.socket.connect((address, port)) @@ -53,61 +55,27 @@ def send(self, content: Union[OscMessage, OscBundle]) -> None: Args: content: Message or bundle to be sent """ - if self.mode == MODE_1_1: - self.socket.sendall(slip.encode(content.dgram)) - else: - b = struct.pack("!I", len(content.dgram)) - self.socket.sendall(b + content.dgram) + self.socket.sendall(self._framing.encode(content.dgram)) def receive(self, timeout: float | None = None) -> List[bytes]: + messages = [] effective_timeout = timeout if timeout is not None else self._timeout self.socket.settimeout(effective_timeout) - if self.mode == MODE_1_1: - try: - buf = self.socket.recv(4096) - except (TimeoutError, socket.timeout): - return [] - if not buf: - return [] - # If the last byte is not an END marker there could be more data coming - while buf[-1] != 192: - try: - newbuf = self.socket.recv(4096) - except (TimeoutError, socket.timeout): - break - if not newbuf: - # Maybe should raise an exception here? - break - buf += newbuf - return [slip.decode(p) for p in buf.split(slip.END_END)] - else: - buf = b"" - try: - lengthbuf = self.socket.recv(4) - except (TimeoutError, socket.timeout): - return [] - (length,) = struct.unpack("!I", lengthbuf) - while length > 0: - try: - newbuf = self.socket.recv(length) - except (TimeoutError, socket.timeout): - return [] - if not newbuf: - return [] - buf += newbuf - length -= len(newbuf) - return [buf] + + with suppress(TimeoutError): + while chunk := self.socket.recv(16384): + messages.extend(self._parser.feed(chunk)) + + return messages def close(self): + self._parser.reset() self.socket.close() class SimpleTCPClient(TCPClient): """Simple OSC client that automatically builds :class:`OscMessage` from arguments""" - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - def send_message( self, address: str, value: Union[ArgValue, Iterable[ArgValue]] = "" ) -> None: @@ -131,7 +99,9 @@ def get_messages(self, timeout: float | None = None) -> Generator: class TCPDispatchClient(SimpleTCPClient): """OSC TCP Client that includes a :class:`Dispatcher` for handling responses and other messages from the server""" - dispatcher = Dispatcher() + def __init__(self, *args: Any, dispatcher: Dispatcher | None = None, **kwargs: Any): + self.dispatcher = dispatcher or Dispatcher() + super().__init__(*args, **kwargs) def handle_messages(self, timeout_sec: float | None = None) -> None: """Wait :int:`timeout` seconds for a message from the server and process each message with the registered @@ -168,16 +138,10 @@ def __init__( """ self.address: str = address self.port: int = port - self.mode: str = mode self.family: socket.AddressFamily = family self._timeout = timeout - - def __await__(self): - async def closure(): - await self.__open__() - return self - - return closure().__await__() + self._framing = get_framing(mode) + self._parser = FramingParser(self._framing) async def __aenter__(self): await self.__open__() @@ -191,83 +155,38 @@ async def __open__(self): async def __aexit__(self, exc_type, exc_val, exc_tb): await self.close() - async def send(self, content: Union[OscMessage, OscBundle]) -> None: + def send(self, content: Union[OscMessage, OscBundle]) -> Awaitable[None]: """Sends an :class:`OscMessage` or :class:`OscBundle` via TCP Args: content: Message or bundle to be sent """ - if self.mode == MODE_1_1: - self.writer.write(slip.encode(content.dgram)) - else: - b = struct.pack("!I", len(content.dgram)) - self.writer.write(b + content.dgram) - await self.writer.drain() + self.writer.write(self._framing.encode(content.dgram)) + return self.writer.drain() + + def _read_with_timeout(self, timeout: float | None) -> Awaitable[bytes]: + return asyncio.wait_for(self.reader.read(16384), timeout) async def receive(self, timeout: float | None = None) -> List[bytes]: + messages = [] effective_timeout = timeout if timeout is not None else self._timeout - if self.mode == MODE_1_1: - try: - buf = await asyncio.wait_for(self.reader.read(4096), effective_timeout) - except (TimeoutError, asyncio.TimeoutError): - return [] - if not buf: - return [] - # If the last byte is not an END marker there could be more data coming - while buf[-1] != 192: - try: - newbuf = await asyncio.wait_for( - self.reader.read(4096), effective_timeout - ) - except (TimeoutError, asyncio.TimeoutError): - break - if not newbuf: - # Maybe should raise an exception here? - break - buf += newbuf - return [slip.decode(p) for p in buf.split(slip.END_END)] - else: - buf = b"" - try: - lengthbuf = await asyncio.wait_for( - self.reader.read(4), effective_timeout - ) - except (TimeoutError, asyncio.TimeoutError): - return [] - - (length,) = struct.unpack("!I", lengthbuf) - while length > 0: - try: - newbuf = await asyncio.wait_for( - self.reader.read(length), effective_timeout - ) - except (TimeoutError, asyncio.TimeoutError): - return [] - if not newbuf: - return [] - buf += newbuf - length -= len(newbuf) - return [buf] - - async def close(self): + + with suppress(TimeoutError): + while chunk := await self._read_with_timeout(effective_timeout): + messages.extend(self._parser.feed(chunk)) + + return messages + + def close(self) -> Awaitable[None]: + self._parser.reset() self.writer.write_eof() self.writer.close() - await self.writer.wait_closed() + return self.writer.wait_closed() class AsyncSimpleTCPClient(AsyncTCPClient): """Simple OSC client that automatically builds :class:`OscMessage` from arguments""" - def __init__( - self, - address: str, - port: int, - family: socket.AddressFamily = socket.AF_INET, - mode: str = MODE_1_1, - timeout: float | None = 30.0, - ): - super().__init__(address, port, family, mode, timeout) - async def send_message( self, address: str, value: Union[ArgValue, Iterable[ArgValue]] = "" ) -> None: @@ -291,7 +210,9 @@ async def get_messages(self, timeout: float | None = None) -> AsyncGenerator: class AsyncDispatchTCPClient(AsyncTCPClient): """OSC Client that includes a :class:`Dispatcher` for handling responses and other messages from the server""" - dispatcher = Dispatcher() + def __init__(self, *args: Any, dispatcher: Dispatcher | None = None, **kwargs: Any): + self.dispatcher = dispatcher or Dispatcher() + super().__init__(*args, **kwargs) async def handle_messages(self, timeout: float | None = None) -> None: """Wait :int:`timeout` seconds for a message from the server and process each message with the registered diff --git a/pythonosc/test/parsing/test_framing.py b/pythonosc/test/parsing/test_framing.py new file mode 100644 index 0000000..72361de --- /dev/null +++ b/pythonosc/test/parsing/test_framing.py @@ -0,0 +1,122 @@ +from pytest import fixture, mark, raises + +from pythonosc.parsing.framing import ( + BinaryLengthFraming, + Framing, + FramingParser, + SLIPFraming, +) + + +class SquareFraming(Framing): + """Framing strategy for ease of testing, wraps everything in [] brackets.""" + + def encode(self, data: bytes) -> bytes: + if data: + return b"[" + data + b"]" + return b"" + + def decode_stream( + self, chunk: bytes | bytearray, start: int = 0 + ) -> tuple[list[bytes], int]: + out = [] + pos = start + + while (a := chunk.find(b"[", pos)) >= 0: + a += 1 + + if (b := chunk.find(b"]", a)) >= 0: + out.append(bytes(chunk[a:b])) + pos = b + 1 + else: + break + + return out, pos - start + + +class CommonFramingTest: + def test_encode_decode(self, framing: Framing) -> None: + messages = [b"test1", b"", b"test2", b"test3"] + expected = [m for m in messages if m] + data = b"".join(framing.encode(message) for message in messages) + + assert framing.decode_stream(data) == (expected, len(data)) + + @mark.parametrize("n,m", [(1, 1), (4, 1), (64, 0)]) + def test_incomplete(self, framing: Framing, n: int, m: int) -> None: + message_a = framing.encode(b"test") + message_b = framing.encode(b"cut here") + data = (message_a + message_b)[:-n] + messages, consumed = framing.decode_stream(data) + + assert messages == messages[:m] + assert data[consumed:] == data[len(message_a) :] + + def test_decode_with_start(self, framing: Framing) -> None: + garbage = b"garbage" + data = garbage + framing.encode(b"test") + + assert framing.decode_stream(data, len(garbage)) == ( + [b"test"], + len(data) - len(garbage), + ) + + +class TestSLIPFraming(CommonFramingTest): + @fixture + def framing(self) -> Framing: + return SLIPFraming() + + def test_encode(self, framing: Framing) -> None: + assert framing.encode(b"te\xc0st") == b"\xc0te\xdb\xdcst\xc0" + + +class TestBinaryLengthFraming(CommonFramingTest): + @fixture + def framing(self) -> Framing: + return BinaryLengthFraming() + + def test_encode(self, framing: Framing) -> None: + assert framing.encode(b"test") == b"\x00\x00\x00\x04test" + + +class TestSquareFraming(CommonFramingTest): + @fixture + def framing(self) -> Framing: + return SquareFraming() + + def test_encode(self, framing: Framing) -> None: + assert framing.encode(b"test") == b"[test]" + + +class TestFramingParser: + @fixture + def parser(self) -> FramingParser: + return FramingParser(SquareFraming(), max_buffer_size=16) + + def test_cold(self, parser: FramingParser) -> None: + assert parser.feed(b"[test]") == [b"test"] + + def test_hot(self, parser: FramingParser) -> None: + assert parser.feed(b"[test") == [] + assert parser.feed(b"123][xxx]") == [b"test123", b"xxx"] + + def test_large_buffer(self, parser: FramingParser) -> None: + parser.feed(b"[test]") + + with raises(ValueError, match="Buffer is too large: 18 > 16"): + parser.feed(b"[test]" * 3) + + def test_compaction(self, parser: FramingParser) -> None: + for _ in range(4): + assert parser.feed(b"[tests]") == [b"tests"] + + def test_reset(self, parser: FramingParser) -> None: + parser.feed(b"[test]") + parser.feed(b"[xxx") + + assert parser.remaining == b"[xxx" + + parser.reset() + + assert parser.remaining == b"" diff --git a/pythonosc/test/test_osc_tcp_server.py b/pythonosc/test/test_osc_tcp_server.py index 0849803..ef5ecaf 100644 --- a/pythonosc/test/test_osc_tcp_server.py +++ b/pythonosc/test/test_osc_tcp_server.py @@ -2,8 +2,11 @@ import unittest import unittest.mock as mock +from pytest import mark, raises + from pythonosc import dispatcher, osc_tcp_server -from pythonosc.slip import END +from pythonosc.parsing.framing import BINARY_LENGTH_FRAMING, SLIP_FRAMING, Framing +from pythonosc.parsing.slip import END _SIMPLE_PARAM_INT_MSG = b"/SYNC\x00\x00\x00,i\x00\x00\x00\x00\x00\x04" @@ -21,15 +24,31 @@ _SIMPLE_MSG_NO_PARAMS_1_1 = END + _SIMPLE_MSG_NO_PARAMS + END +class TestGetFraming: + @mark.parametrize( + ["mode", "framing"], + [ + (osc_tcp_server.MODE_1_0, BINARY_LENGTH_FRAMING), + (osc_tcp_server.MODE_1_1, SLIP_FRAMING), + ], + ) + def test_ok(self, mode: str, framing: Framing) -> None: + assert osc_tcp_server.get_framing(mode) is framing + + def test_unknown(self) -> None: + with raises(ValueError, match=r"Unsupported OSC mode: test"): + osc_tcp_server.get_framing("test") + + class TestTCP_1_1_Handler(unittest.TestCase): def setUp(self): super().setUp() self.dispatcher = dispatcher.Dispatcher() # We do not want to create real UDP connections during unit tests. - self.server = unittest.mock.Mock(spec=osc_tcp_server.BlockingOSCTCPServer) - # Need to attach property mocks to types, not objects... weird. - type(self.server).dispatcher = unittest.mock.PropertyMock( - return_value=self.dispatcher + self.server = unittest.mock.Mock( + spec=osc_tcp_server.BlockingOSCTCPServer, + dispatcher=self.dispatcher, + framing=SLIP_FRAMING, ) self.client_address = ("127.0.0.1", 8080) self.mock_meth = unittest.mock.MagicMock() @@ -44,7 +63,7 @@ def test_no_match(self): _SIMPLE_PARAM_INT_MSG_1_1, b"", ] - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.assertFalse(self.mock_meth.called) def test_match_with_args(self): @@ -52,7 +71,7 @@ def test_match_with_args(self): mock_sock = mock.Mock() mock_sock.recv = mock.Mock() mock_sock.recv.side_effect = [_SIMPLE_PARAM_INT_MSG_1_1, b""] - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC", [1, 2, 3], 4) def test_match_int9(self): @@ -60,7 +79,7 @@ def test_match_int9(self): mock_sock = mock.Mock() mock_sock.recv = mock.Mock() mock_sock.recv.side_effect = [_SIMPLE_PARAM_INT_9_1_1, b""] - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.assertTrue(self.mock_meth.called) self.mock_meth.assert_called_with("/debug", 9) @@ -69,7 +88,7 @@ def test_match_without_args(self): mock_sock = mock.Mock() mock_sock.recv = mock.Mock() mock_sock.recv.side_effect = [_SIMPLE_MSG_NO_PARAMS_1_1, b""] - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC") def test_match_default_handler(self): @@ -77,7 +96,7 @@ def test_match_default_handler(self): mock_sock = mock.Mock() mock_sock.recv = mock.Mock() mock_sock.recv.side_effect = [_SIMPLE_MSG_NO_PARAMS_1_1, b""] - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC") def test_response_no_args(self): @@ -90,7 +109,7 @@ def respond(*args, **kwargs): mock_sock.recv.side_effect = [_SIMPLE_MSG_NO_PARAMS_1_1, b""] mock_sock.sendall = mock.Mock() mock_sock.sendall.return_value = None - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) mock_sock.sendall.assert_called_with(b"\xc0/SYNC\00\00\00,\00\00\00\xc0") def test_response_with_args(self): @@ -108,7 +127,7 @@ def respond(*args, **kwargs): mock_sock.recv.side_effect = [_SIMPLE_MSG_NO_PARAMS_1_1, b""] mock_sock.sendall = mock.Mock() mock_sock.sendall.return_value = None - osc_tcp_server._TCPHandler1_1(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) mock_sock.sendall.assert_called_with( b"\xc0/SYNC\00\00\00,isf\x00\x00\x00\x00\x00\x00\x00\x012\x00\x00\x00@@\x00\x00\xc0" ) @@ -119,10 +138,10 @@ def setUp(self): super().setUp() self.dispatcher = dispatcher.Dispatcher() # We do not want to create real UDP connections during unit tests. - self.server = unittest.mock.Mock(spec=osc_tcp_server.BlockingOSCTCPServer) - # Need to attach property mocks to types, not objects... weird. - type(self.server).dispatcher = unittest.mock.PropertyMock( - return_value=self.dispatcher + self.server = unittest.mock.Mock( + spec=osc_tcp_server.BlockingOSCTCPServer, + dispatcher=self.dispatcher, + framing=BINARY_LENGTH_FRAMING, ) self.client_address = ("127.0.0.1", 8080) self.mock_meth = unittest.mock.MagicMock() @@ -139,7 +158,7 @@ def test_no_match(self): _SIMPLE_PARAM_INT_MSG, b"", ] - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.assertFalse(self.mock_meth.called) def test_match_with_args(self): @@ -151,7 +170,7 @@ def test_match_with_args(self): _SIMPLE_PARAM_INT_MSG, b"", ] - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC", [1, 2, 3], 4) def test_match_int9(self): @@ -159,7 +178,7 @@ def test_match_int9(self): mock_sock = mock.Mock() mock_sock.recv = mock.Mock() mock_sock.recv.side_effect = [LEN_SIMPLE_PARAM_INT_9, _SIMPLE_PARAM_INT_9, b""] - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.assertTrue(self.mock_meth.called) self.mock_meth.assert_called_with("/debug", 9) @@ -172,7 +191,7 @@ def test_match_without_args(self): _SIMPLE_MSG_NO_PARAMS, b"", ] - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC") def test_match_default_handler(self): @@ -184,7 +203,7 @@ def test_match_default_handler(self): _SIMPLE_MSG_NO_PARAMS, b"", ] - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) self.mock_meth.assert_called_with("/SYNC") def test_response_no_args(self): @@ -201,7 +220,7 @@ def respond(*args, **kwargs): ] mock_sock.sendall = mock.Mock() mock_sock.sendall.return_value = None - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) mock_sock.sendall.assert_called_with( b"\x00\x00\x00\x0c/SYNC\00\00\00,\00\00\00" ) @@ -225,7 +244,7 @@ def respond(*args, **kwargs): ] mock_sock.sendall = mock.Mock() mock_sock.sendall.return_value = None - osc_tcp_server._TCPHandler1_0(mock_sock, self.client_address, self.server) + osc_tcp_server.TCPHandler(mock_sock, self.client_address, self.server) mock_sock.sendall.assert_called_with( b"\x00\x00\x00\x1c/SYNC\00\00\00,isf\x00\x00\x00\x00\x00\x00\x00\x012\x00\x00\x00@@\x00\x00" ) @@ -236,10 +255,9 @@ def setUp(self): super().setUp() self.dispatcher = dispatcher.Dispatcher() # We do not want to create real UDP connections during unit tests. - self.server = unittest.mock.Mock(spec=osc_tcp_server.BlockingOSCTCPServer) - # Need to attach property mocks to types, not objects... weird. - type(self.server).dispatcher = unittest.mock.PropertyMock( - return_value=self.dispatcher + self.server = unittest.mock.Mock( + spec=osc_tcp_server.BlockingOSCTCPServer, + dispatcher=self.dispatcher, ) self.client_address = ("127.0.0.1", 8080) self.mock_writer = mock.Mock() diff --git a/pythonosc/test/test_tcp_client.py b/pythonosc/test/test_tcp_client.py index a0089f4..21e9f01 100644 --- a/pythonosc/test/test_tcp_client.py +++ b/pythonosc/test/test_tcp_client.py @@ -2,7 +2,8 @@ import unittest from unittest import mock -from pythonosc import osc_message_builder, slip, tcp_client +from pythonosc import osc_message_builder, tcp_client +from pythonosc.parsing import slip class TestTcpClient(unittest.TestCase):