Coverage for /home/fedora/jumpstarter/packages/jumpstarter/jumpstarter/streams/blocking.py: 82%

1from dataclasses import dataclass 

2 

3from anyio.abc import AnyByteStream 

4from anyio.from_thread import BlockingPortal 

5 

6 

7@dataclass(kw_only=True) 

8class BlockingStream: 

9 stream: AnyByteStream 

10 portal: BlockingPortal 

11 

12 def send(self, data: bytes) -> None: 

13 return self.portal.call(self.stream.send, data) 

14 

15 def receive(self) -> bytes: 

16 return self.portal.call(self.stream.receive)