Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions alsa_midi/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -404,6 +404,7 @@ def close(self):
self._handle_p = None # type: ignore
self.handle = None # type: ignore
self._event_parser = None
self._fd = -1

def _get_fds(self):
"""Get the file descriptor number for the client connection into :data:`self._fd`."""
Expand All @@ -430,6 +431,21 @@ def _get_event_parser(self):
self._event_parser = parser
return parser

def fileno(self):
"""Return file descriptor assigned to this connection.

This file descriptor can be used with select, poll, or other event
loop mechanisms for monitoring ALSA sequencer events.

Wraps :alsa:`snd_seq_poll_descriptors`.

:return: file descriptor number
:raises OSError: if the connection is closed or file descriptor is not available
"""
if self._fd >= 0:
return self._fd
raise OSError("File descriptor not available")

def get_sequencer_name(self) -> str:
"""Get sequencer name.

Expand Down
40 changes: 40 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -611,3 +611,43 @@ def test_set_client_pool_input():
client.set_client_pool_input(200)
pool = client.get_client_pool()
assert pool.input_pool == 200


@pytest.mark.require_alsa_seq
def test_fileno():
client = SequencerClient("test")
fd = client.fileno()
assert isinstance(fd, int)
assert fd >= 0
client.close()


@pytest.mark.require_alsa_seq
def test_fileno_after_close():
client = SequencerClient("test")
fd = client.fileno()
assert fd >= 0
client.close()
with pytest.raises(OSError) as exc_info:
client.fileno()
assert "File descriptor not available" in str(exc_info.value)


@pytest.mark.require_alsa_seq
def test_fileno_async_client():
client = AsyncSequencerClient("test")
fd = client.fileno()
assert isinstance(fd, int)
assert fd >= 0
client.close()


@pytest.mark.require_alsa_seq
def test_fileno_async_client_after_close():
client = AsyncSequencerClient("test")
fd = client.fileno()
assert fd >= 0
client.close()
with pytest.raises(OSError) as exc_info:
client.fileno()
assert "File descriptor not available" in str(exc_info.value)
Loading