Skip to content

Conversation

@andrewleech
Copy link
Contributor

Returns two bi-directional linked stream objects where writes to one can be read from the other and vice/versa.

This can be used somewhat similarly to a socket.socketpair in python, like a pipe of data that can be used to connect stream consumers (eg. asyncio.StreamWriter, mock Uart)

@andrewleech andrewleech changed the title Draft: micropython/streampair: Package to create bi-directional linked stream objects. micropython/streampair: Package to create bi-directional linked stream objects. Aug 13, 2024
@andrewleech
Copy link
Contributor Author

The RingIO module has been merged in micropython so this should be updated to use it

@andrewleech
Copy link
Contributor Author

andrewleech commented Oct 30, 2025

Updated this PR with the following changes:

API Updates:

  • Replace deprecated ringbuffer with RingIO name (they're the same underlying C type, but RingIO is the documented API name)
  • Remove unused collections.deque import

New Tests:
Added two tests that demonstrate and verify the ioctl protocol implementation works correctly with select.poll() and asyncio:

  1. test_select_poll_compatibility() - Verifies that StreamPair objects can be registered with select.poll() and correctly report POLLIN events when data is available

  2. test_streamreader_direct_usage() - Demonstrates that StreamPair works directly with asyncio.StreamReader without needing any wrapper or special handling

Technical Detail:

The existing ioctl implementation already correctly returns -1 (_MP_STREAM_ERROR) for unsupported operations.
This is the protocol requirement that enables select.poll() to work with pure Python stream objects.

Since asyncio.StreamReader is built on top of select.poll(), this means StreamPair (and any Python object with proper ioctl implementation) works seamlessly with async I/O.

These tests serve as documentation that pure Python stream objects don't require C-level integration to work with micropython's polling and async infrastructure - they just need to implement the ioctl protocol correctly.

- Replace deprecated ringbuffer with RingIO name.
- Remove unused collections.deque import.
- Add test_select_poll_compatibility() to verify select.poll().
- Add test_streamreader_direct_usage() for asyncio.StreamReader.

The existing ioctl implementation correctly returns -1 for
unsupported operations, enabling select.poll() to work with pure
Python stream objects. Since asyncio.StreamReader is built on
select.poll(), StreamPair works seamlessly with async I/O.

These tests document that Python objects with proper ioctl()
implementation work with micropython polling and async infrastructure
without requiring C-level integration.

Signed-off-by: Andrew Leech <[email protected]>
Signed-off-by: Andrew Leech <[email protected]>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants