Add reactor.signal_write_ready(fd) API

Provides a clean API for protocols to signal async completion instead
of manually sending to the reactor PID. The reactor PID is stored per-fd
in init_connection and cleaned up in close_connection.

Author: benoitc

priv/_erlang_impl/_reactor.py

@@ -47,6 +47,7 @@ def write_ready(self):
     'on_read_ready',
     'on_write_ready',
     'close_connection',
+    'signal_write_ready',
 ]
 
 
@@ -153,6 +154,7 @@ def write(self, data: bytes) -> int:
 # =============================================================================
 
 _protocols: Dict[int, Protocol] = {}
+_reactor_pids: Dict[int, object] = {}  # fd -> reactor PID
 _protocol_factory: Optional[Callable[[], Protocol]] = None
 
 
@@ -193,11 +195,15 @@ def init_connection(fd: int, client_info: dict):
         fd: File descriptor
         client_info: Connection metadata from Erlang
     """
-    global _protocols, _protocol_factory
+    global _protocols, _protocol_factory, _reactor_pids
     if _protocol_factory is not None:
         proto = _protocol_factory()
         proto.connection_made(fd, client_info)
         _protocols[fd] = proto
+        # Store reactor PID for signal_write_ready
+        reactor_pid = client_info.get('reactor_pid')
+        if reactor_pid is not None:
+            _reactor_pids[fd] = reactor_pid
 
 
 def on_read_ready(fd: int) -> str:
@@ -246,6 +252,27 @@ def close_connection(fd: int):
         fd: File descriptor
     """
     proto = _protocols.pop(fd, None)
+    _reactor_pids.pop(fd, None)
     if proto is not None:
         proto.closed = True
         proto.connection_lost()
+
+
+def signal_write_ready(fd: int) -> bool:
+    """Signal the reactor that a response is ready for the given fd.
+
+    Call this after an async task completes and the response buffer is ready.
+    The reactor will then trigger write selection for the fd.
+
+    Args:
+        fd: File descriptor with pending response
+
+    Returns:
+        True if signal was sent, False if no reactor PID registered
+    """
+    import erlang
+    reactor_pid = _reactor_pids.get(fd)
+    if reactor_pid is not None:
+        erlang.send(reactor_pid, ('write_ready', fd))
+        return True
+    return False

test/py_reactor_SUITE.erl

@@ -215,35 +215,20 @@ _close_test_result = run_close_test()
 async_pending_test(_Config) ->
     %% Protocol factory code to run in reactor context
     SetupCode = <<"
-import erlang
 import erlang.reactor as reactor
 
 class AsyncPendingProtocol(reactor.Protocol):
     '''Protocol that returns async_pending and signals completion.'''
 
     def __init__(self):
         super().__init__()
-        self.reactor_pid = None
         self.pending_response = b''
 
-    def connection_made(self, fd, client_info):
-        super().connection_made(fd, client_info)
-        self.reactor_pid = client_info.get('reactor_pid')
-
     def data_received(self, data):
-        import sys
         self.pending_response = b'ASYNC:' + data
         # Immediately complete the task and signal reactor
         self.write_buffer.extend(self.pending_response)
-        if self.reactor_pid:
-            print(f'Sending write_ready to {self.reactor_pid} for fd={self.fd}', file=sys.stderr)
-            try:
-                erlang.send(self.reactor_pid, ('write_ready', self.fd))
-                print('write_ready sent successfully', file=sys.stderr)
-            except Exception as e:
-                print(f'erlang.send failed: {e}', file=sys.stderr)
-        else:
-            print('No reactor_pid available!', file=sys.stderr)
+        reactor.signal_write_ready(self.fd)
         return 'async_pending'
 
     def write_ready(self):