61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from threading import Lock
|
|
from typing import Callable, List
|
|
|
|
from .io_buffer import IOBuffer
|
|
|
|
class WebIOBuffer(IOBuffer):
|
|
"""
|
|
Thread-safe WebIOBuffer that maintains the synchronous IOBuffer interface.
|
|
Uses threading primitives instead of asyncio for synchronization.
|
|
"""
|
|
def __init__(self):
|
|
self._stdin_buffer = ""
|
|
self._stdout_buffer = ""
|
|
self._stdout_change_handlers: List[Callable[[str], None]] = []
|
|
|
|
def add_stdout_change_handler(self, handler: Callable[[str], None]) -> None:
|
|
"""
|
|
Add a callback for stdout changes.
|
|
"""
|
|
if handler not in self._stdout_change_handlers:
|
|
self._stdout_change_handlers.append(handler)
|
|
|
|
def read(self) -> str:
|
|
"""Thread-safe read from stdin buffer."""
|
|
content = self._stdin_buffer
|
|
self._stdin_buffer = ""
|
|
return content
|
|
|
|
def write(self, content: str) -> None:
|
|
"""Thread-safe write to stdout buffer."""
|
|
self._stdout_buffer += content
|
|
for handler in self._stdout_change_handlers:
|
|
handler(self._stdout_buffer)
|
|
|
|
def buffer_length(self) -> int:
|
|
"""
|
|
Get the current length of the stdin buffer.
|
|
|
|
Returns:
|
|
int: Number of characters in the stdin buffer
|
|
"""
|
|
return len(self._stdin_buffer)
|
|
|
|
def append_stdin(self, content: str) -> None:
|
|
"""
|
|
Append content to the stdin buffer.
|
|
|
|
Args:
|
|
content: String content to append to the stdin buffer
|
|
"""
|
|
self._stdin_buffer += content
|
|
|
|
def get_stdout(self) -> str:
|
|
"""Thread-safe get stdout content."""
|
|
return self._stdout_buffer
|
|
|
|
def clear_stdout(self) -> None:
|
|
"""Thread-safe clear stdout buffer."""
|
|
self._stdout_buffer = ""
|
|
for handler in self._stdout_change_handlers:
|
|
handler(self._stdout_buffer) |