Pyjo.IOLoop.Stream - Non-blocking I/O stream

from Pyjo.IOLoop.Stream import Stream

# Create stream
stream = Stream(handle)
stream.on('read', lambda stream, chunk: ...)
stream.on('close', lambda stream: ...)
stream.on('error', lambda stream, err: ...)

# Or import the module, as in Google Python Style Guide

import Pyjo.IOLoop.Stream

# Create stream
stream = Pyjo.IOLoop.Stream.new(handle)
stream.on('read', lambda stream, chunk: ...)
stream.on('close', lambda stream: ...)
stream.on('error', lambda stream, err: ...)

# Start and stop watching for new data
stream.start()
stream.stop()

# Start reactor if necessary
if not stream.reactor.is_running:
    stream.reactor.start()

Pyjo.IOLoop.Stream is a container for I/O streams used by Pyjo.IOLoop. A handle is a socket, an ssl.SSLSocket too, or anything else with a file descriptor, like a pipe.

The end of a with block closes the stream.

Events

Pyjo.IOLoop.Stream inherits all events from Pyjo.EventEmitter and can emit the following new ones.

close

stream.on('close', lambda stream: ...)

Emitted if the stream gets closed.

drain

stream.on('drain', lambda stream: ...)

Emitted once all data has been written.

error

stream.on('error', lambda stream, err: ...)

Emitted if an error occurs on the stream, fatal if unhandled. The error is the OSError of the handle.

read

stream.on('read', lambda stream, chunk: ...)

Emitted if new data arrives on the stream.

timeout

stream.on('timeout', lambda stream: ...)

Emitted if the stream has been inactive for too long and will get closed automatically.

write

stream.on('write', lambda stream, chunk: ...)

Emitted if new data has been written to the stream.

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::IOLoop::Stream of Mojolicious 9.49.

class Pyjo.IOLoop.Stream.Pyjo_IOLoop_Stream(handle: Any = None)

Pyjo.IOLoop.Stream inherits all methods from Pyjo.EventEmitter and implements the following new ones.

property bytes_read: int

num = stream.bytes_read

Number of bytes received.

property bytes_waiting: int

num = stream.bytes_waiting

Number of bytes that have been enqueued with write() and are waiting to be written.

property bytes_written: int

num = stream.bytes_written

Number of bytes written.

property can_write: bool

boolean = stream.can_write

True if calling write() is safe.

close() → None

stream.close()

Close stream immediately.

close_gracefully() → None

stream.close_gracefully()

Close stream gracefully.

property handle: Any

handle = stream.handle

Get handle for stream, usually a socket.socket or ssl.SSLSocket object.

high_water_mark

size = stream.high_water_mark
stream.high_water_mark = 1024

Maximum size of write() buffer in bytes before can_write is false, defaults to 1048576 (1MiB).

property is_readable: bool

boolean = stream.is_readable

Quick non-blocking check if stream is readable, useful for identifying tainted sockets.

property is_writing: bool

boolean = stream.is_writing

Check if stream is writing.

property reactor: Pyjo_Reactor

reactor = stream.reactor
stream.reactor = Pyjo.Reactor.Poll.new()

Low-level event reactor, defaults to the reactor attribute value of the global Pyjo.IOLoop singleton. Note that this attribute is weakened.

start() → None

stream.start()

Start or resume watching for new data on the stream.

steal_handle() → Any

handle = stream.steal_handle()

Steal handle and prevent it from getting closed automatically.

stop() → None

stream.stop()

Stop watching for new data on the stream.

property timeout: float

timeout = stream.timeout
stream.timeout = 45

Maximum amount of time in seconds stream can be inactive before getting closed automatically, defaults to 15. Setting the value to 0 will allow this stream to be inactive indefinitely.

write(chunk: bytes, cb: DrainCallback | None = None) → Self

stream = stream.write(chunk)
stream = stream.write(chunk, lambda stream: ...)

Enqueue data to be written to the stream as soon as possible, the optional drain callback will be executed once all data has been written.