### Custom RTMP Controller Implementation

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Implement a custom controller to handle RTMP sessions, including handshake and message processing. This example uses pyrtmp's SessionManager to iterate over incoming chunks and convert them into messages.

```python
async def my_controller(reader, writer):
    from pyrtmp.messages import SessionManager
    session = SessionManager(reader=reader, writer=writer)
    await session.handshake()
    async for chunk in session.read_chunks_from_stream():
        message = chunk.as_message()
        # Handle messages...
```

--------------------------------

### Initialize RTMPT Server Wrapper

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Sets up a basic Quart application for tunneling RTMP over HTTP. Requires the RTMPTWrapper utility.

```python
import asyncio
from quart import Quart, request
import quart
import uuid
from pyrtmp.misc.rtmpt import RTMPTWrapper

app = Quart(__name__)
sessions = {}
```

--------------------------------

### Production RTMP Server with Multi-Process and uvloop

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Sets up an RTMP server for production using multiple processes and uvloop for enhanced performance. Includes a worker function to run the server and a main block to manage processes.

```python
import asyncio
from multiprocessing import Process
from pyrtmp import RTMPProtocol
from pyrtmp.messages import SessionManager

async def production_controller(reader, writer):
    session = SessionManager(reader=reader, writer=writer)
    # Your controller implementation...

async def serve_rtmp(port: int):
    loop = asyncio.get_running_loop()
    server = await loop.create_server(
        lambda: RTMPProtocol(controller=production_controller, loop=loop),
        '0.0.0.0',
        port
    )
    async with server:
        await server.serve_forever()

def worker(port: int):
    asyncio.run(serve_rtmp(port))

if __name__ == "__main__":
    IS_DEBUG = False
    NUM_PROCESS = 4  # Match CPU count

    if IS_DEBUG:
        # Single process for debugging
        asyncio.run(serve_rtmp(1935))
    else:
        # Multi-process with uvloop for production
        import uvloop
        uvloop.install()

        processes = []
        for i in range(NUM_PROCESS):
            port = 1936 + i  # Ports 1936, 1937, 1938, 1939
            p = Process(target=worker, args=(port,)) 
            p.start()
            processes.append(p)

        for p in processes:
            p.join()
```

--------------------------------

### Implement RTMP Server with RTMPProtocol

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Use RTMPProtocol to integrate with the asyncio event loop for server creation. This approach is recommended for production environments.

```python
import asyncio
from pyrtmp import RTMPProtocol, StreamClosedException
from pyrtmp.messages import SessionManager

async def my_controller(reader, writer):
    session = SessionManager(reader=reader, writer=writer)
    try:
        await session.handshake()
        async for chunk in session.read_chunks_from_stream():
            message = chunk.as_message()
            # Process messages...
    except StreamClosedException:
        pass

async def serve_rtmp():
    loop = asyncio.get_running_loop()

    # Method 1: Using RTMPProtocol (recommended for production)
    server = await loop.create_server(
        lambda: RTMPProtocol(controller=my_controller, loop=loop),
        '0.0.0.0',
        1935
    )

    # Method 2: Using asyncio.start_server (simpler alternative)
    # server = await asyncio.start_server(my_controller, '0.0.0.0', 1935)

    addr = server.sockets[0].getsockname()
    print(f'RTMP Server running on {addr}')

    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(serve_rtmp())
```

--------------------------------

### Test RTMP Server with FFmpeg

Source: https://github.com/aungyezawdev/pyrtmp/blob/main/README.md

Use this FFmpeg command to send a test stream to your RTMP server configured on port 1935. Replace 'my_test_file.flv' with your actual input file.

```bash
ffmpeg -i my_test_file.flv -c:v copy -c:a copy -f flv rtmp://127.0.0.1:1935/test/sample
```

--------------------------------

### Simple RTMP Controller for Receiving and Saving Streams

Source: https://github.com/aungyezawdev/pyrtmp/blob/main/README.md

Implement a controller to manage RTMP connections, handshake, and stream data. This snippet handles connection, publishing, metadata, video, and audio messages, saving them to an FLV file in the temporary directory. Ensure the necessary imports are present.

```python
import asyncio
import logging
import os
import tempfile

from pyrtmp import StreamClosedException, RTMPProtocol
from pyrtmp.messages import SessionManager
from pyrtmp.messages.audio import AudioMessage
from pyrtmp.messages.command import NCConnect, NCCreateStream, NSPublish, NSCloseStream, NSDeleteStream
from pyrtmp.messages.data import MetaDataMessage
from pyrtmp.messages.protocolcontrol import WindowAcknowledgementSize, SetChunkSize, SetPeerBandwidth
from pyrtmp.messages.usercontrol import StreamBegin
from pyrtmp.messages.video import VideoMessage
from pyrtmp.misc.flvdump import FLVFile, FLVMediaType

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


async def simple_controller(reader, writer):
    session = SessionManager(reader=reader, writer=writer)
    flv = None
    try:
        logger.debug(f'Client connected {session.peername}')

        # do handshake
        await session.handshake()

        # read chunks
        async for chunk in session.read_chunks_from_stream():
            message = chunk.as_message()
            logger.debug(f"Receiving {str(message)} {message.chunk_id}")
            if isinstance(message, NCConnect):
                session.write_chunk_to_stream(WindowAcknowledgementSize(ack_window_size=5000000))
                session.write_chunk_to_stream(SetPeerBandwidth(ack_window_size=5000000, limit_type=2))
                session.write_chunk_to_stream(StreamBegin(stream_id=0))
                session.write_chunk_to_stream(SetChunkSize(chunk_size=8192))
                session.writer_chunk_size = 8192
                session.write_chunk_to_stream(message.create_response())
                await session.drain()
                logger.debug("Response to NCConnect")
            elif isinstance(message, WindowAcknowledgementSize):
                pass
            elif isinstance(message, NCCreateStream):
                session.write_chunk_to_stream(message.create_response())
                await session.drain()
                logger.debug("Response to NCCreateStream")
            elif isinstance(message, NSPublish):
                # create flv file at temp
                flv = FLVFile(os.path.join(tempfile.gettempdir(), message.publishing_name))
                session.write_chunk_to_stream(StreamBegin(stream_id=1))
                session.write_chunk_to_stream(message.create_response())
                await session.drain()
                logger.debug("Response to NSPublish")
            elif isinstance(message, MetaDataMessage):
                # Write meta data to file
                flv.write(0, message.to_raw_meta(), FLVMediaType.OBJECT)
            elif isinstance(message, SetChunkSize):
                session.reader_chunk_size = message.chunk_size
            elif isinstance(message, VideoMessage):
                # Write video data to file
                flv.write(message.timestamp, message.payload, FLVMediaType.VIDEO)
            elif isinstance(message, AudioMessage):
                # Write data data to file
                flv.write(message.timestamp, message.payload, FLVMediaType.AUDIO)
            elif isinstance(message, NSCloseStream):
                pass
            elif isinstance(message, NSDeleteStream):
                pass
            else:
                logger.debug(f"Unknown message {str(message)}")

    except StreamClosedException as ex:
        logger.debug(f"Client {session.peername} disconnected!")
    finally:
        if flv:
            flv.close()


async def serve_rtmp(use_protocol=True):
    loop = asyncio.get_running_loop()
    if use_protocol is True:
        server = await loop.create_server(lambda: RTMPProtocol(controller=simple_controller, loop=loop), '0.0.0.0', 1935)
    else:
        server = await asyncio.start_server(simple_controller, '0.0.0.0', 1935)
    addr = server.sockets[0].getsockname()
    logger.info(f'Serving on {addr}')
    async with server:
        await server.serve_forever()

def wrapper(port: int):
    asyncio.run(serve_rtmp(port=port))

    
IS_DEBUG=True
NUM_PROCESS=2

if __name__ == "__main__":
    if IS_DEBUG is True:
        wrapper(1935)
    else:
        from multiprocessing import Process
        import uvloop
        uvloop.install()
        process = []
        for i in range(NUM_PROCESS):

```

--------------------------------

### Handle NetConnection Commands

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Processes NCConnect and NCCreateStream messages to establish client connections and initialize message streams.

```python
from pyrtmp.messages.command import NCConnect, NCCreateStream
from pyrtmp.messages.protocolcontrol import WindowAcknowledgementSize, SetChunkSize, SetPeerBandwidth
from pyrtmp.messages.usercontrol import StreamBegin

async def handle_connection(session, message):
    if isinstance(message, NCConnect):
        # Access connection details from command object
        print(f"App: {message.command_object.get('app')}")
        print(f"Flash Ver: {message.command_object.get('flashVer')}")
        print(f"TC URL: {message.command_object.get('tcUrl')}")
        print(f"Transaction ID: {message.transaction_id}")

        # Send required protocol control messages
        session.write_chunk_to_stream(WindowAcknowledgementSize(ack_window_size=5000000))
        session.write_chunk_to_stream(SetPeerBandwidth(ack_window_size=5000000, limit_type=2))
        session.write_chunk_to_stream(StreamBegin(stream_id=0))
        session.write_chunk_to_stream(SetChunkSize(chunk_size=8192))
        session.writer_chunk_size = 8192

        # Send success response (creates _result message)
        session.write_chunk_to_stream(message.create_response())
        await session.drain()

    elif isinstance(message, NCCreateStream):
        # Client requesting new stream for publishing
        print(f"CreateStream transaction: {message.transaction_id}")

        # Send stream creation response
        session.write_chunk_to_stream(message.create_response())
        await session.drain()
```

--------------------------------

### RTMPT Session Initialization Endpoint

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

An endpoint to initialize a new RTMPT session using Quart. It generates a session ID, stores the session wrapper, and returns the session ID with specific headers.

```python
@app.route('/open/<int:segment>', methods=['POST'])
async def rtmpt_open(segment: int):
    """Initialize new RTMPT session"""
    sid = uuid.uuid4().hex
    sessions[sid] = RTMPTWrapper(
        session_id=sid,
        peer=request.scope["client"],
        controller=my_controller,
        loop=asyncio.get_running_loop()
    )
    resp = quart.Response(sid)
    resp.headers['Content-Type'] = 'application/x-fcs'
    resp.headers['Cache-Control'] = 'no-cache'
    return resp
```

--------------------------------

### Handle and Configure Protocol Control Messages

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Manages RTMP connection parameters like chunk size and acknowledgement windows. Requires importing specific message classes from pyrtmp.messages.protocolcontrol.

```python
from pyrtmp.messages.protocolcontrol import (
    SetChunkSize,
    WindowAcknowledgementSize,
    SetPeerBandwidth,
    Acknowledgement,
    AbortMessage
)

async def handle_protocol_control(session, message):
    if isinstance(message, SetChunkSize):
        # Client changed chunk size - update reader accordingly
        # Chunk size can range from 1 to 2147483647 bytes
        session.reader_chunk_size = message.chunk_size
        print(f"Chunk size updated to: {message.chunk_size}")

    elif isinstance(message, WindowAcknowledgementSize):
        # Client requesting acknowledgement window
        print(f"Window ack size: {message.ack_window_size}")

# Sending protocol control messages to client
def configure_connection(session):
    # Set acknowledgement window size (bytes before ack required)
    session.write_chunk_to_stream(
        WindowAcknowledgementSize(ack_window_size=5000000)
    )

    # Set peer bandwidth with limit type
    # limit_type: 0=Hard, 1=Soft, 2=Dynamic
    session.write_chunk_to_stream(
        SetPeerBandwidth(ack_window_size=5000000, limit_type=2)
    )

    # Set chunk size for outgoing messages
    session.write_chunk_to_stream(
        SetChunkSize(chunk_size=8192)
    )
    session.writer_chunk_size = 8192

    # Send acknowledgement for received bytes
    session.write_chunk_to_stream(
        Acknowledgement(seq_number=session.total_read_bytes)
    )
```

--------------------------------

### Handle NetStream Commands

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Manages stream lifecycle events such as publishing, closing, and deleting streams.

```python
from pyrtmp.messages.command import NSPublish, NSCloseStream, NSDeleteStream
from pyrtmp.messages.usercontrol import StreamBegin

async def handle_stream_commands(session, message):
    if isinstance(message, NSPublish):
        # Client starting to publish a stream
        print(f"Publishing stream: {message.publishing_name}")
        print(f"Publish type: {message.publishing_type}")  # "live", "record", or "append"
        print(f"Stream ID: {message.msg_stream_id}")

        # Signal stream beginning and send publish start response
        session.write_chunk_to_stream(StreamBegin(stream_id=1))
        session.write_chunk_to_stream(message.create_response())
        await session.drain()

        # Now ready to receive audio/video data

    elif isinstance(message, NSCloseStream):
        # Client closing the stream
        print("Stream closed by client")

    elif isinstance(message, NSDeleteStream):
        # Client deleting the stream
        print(f"Stream deleted: {message.stream_id}")
```

--------------------------------

### Record RTMP Streams to FLV

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Uses FLVFile to capture and persist incoming audio, video, and metadata messages into an FLV container.

```python
import os
import tempfile
from pyrtmp.misc.flvdump import FLVFile, FLVMediaType
from pyrtmp.messages.data import MetaDataMessage
from pyrtmp.messages.video import VideoMessage
from pyrtmp.messages.audio import AudioMessage

async def record_stream(session, publishing_name):
    # Create FLV file for recording
    output_path = os.path.join(tempfile.gettempdir(), f"{publishing_name}.flv")
    flv = FLVFile(output_path)

    try:
        async for chunk in session.read_chunks_from_stream():
            message = chunk.as_message()

            if isinstance(message, MetaDataMessage):
                # Write stream metadata (codec info, dimensions, etc.)
                # to_raw_meta() serializes metadata in FLV-compatible format
                flv.write(0, message.to_raw_meta(), FLVMediaType.OBJECT)
                print(f"Metadata: {message.meta}")

            elif isinstance(message, VideoMessage):
                # Write video frame with timestamp
                flv.write(message.timestamp, message.payload, FLVMediaType.VIDEO)

            elif isinstance(message, AudioMessage):
                # Write audio frame with timestamp
                flv.write(message.timestamp, message.payload, FLVMediaType.AUDIO)

    finally:
        # Always close the FLV file to flush remaining data
        flv.close()
        print(f"Recording saved to: {output_path}")
```

--------------------------------

### Nginx Stream Load Balancing Configuration

Source: https://github.com/aungyezawdev/pyrtmp/blob/main/README.md

Configure Nginx to load balance RTMP traffic across multiple backend servers. Ensure the 'stream' module is enabled in your Nginx build.

```nginx
stream {

    upstream stream_backend {
        127.0.0.1:1936;
        127.0.0.1:1937;
    }

    server {
        listen     1935;
        proxy_pass stream_backend;
    }
}
```

--------------------------------

### Manage User Control Messages and Stream Events

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Signals stream state changes and handles ping/pong synchronization. Uses classes from pyrtmp.messages.usercontrol.

```python
from pyrtmp.messages.usercontrol import (
    StreamBegin,
    StreamEOF,
    StreamDry,
    StreamIsRecorded,
    SetBufferLength,
    PingRequest,
    PingResponse
)

def send_stream_events(session, stream_id):
    # Signal that a stream has started
    session.write_chunk_to_stream(StreamBegin(stream_id=stream_id))

    # Signal that a stream has ended
    session.write_chunk_to_stream(StreamEOF(stream_id=stream_id))

    # Signal that stream has no more data
    session.write_chunk_to_stream(StreamDry(stream_id=stream_id))

    # Signal that stream is being recorded
    session.write_chunk_to_stream(StreamIsRecorded(stream_id=stream_id))

    # Set client buffer length in milliseconds
    session.write_chunk_to_stream(
        SetBufferLength(stream_id=stream_id, milliseconds=3000)
    )

async def handle_ping(session, message):
    if isinstance(message, PingRequest):
        # Respond to ping with matching timestamp
        session.write_chunk_to_stream(
            PingResponse(timestamp=message.timestamp)
        )
        await session.drain()
```

--------------------------------

### Handle RTMP Sessions with SessionManager

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

Use SessionManager to perform handshakes and process incoming RTMP chunks. This class manages the connection state and provides an async iterator for message handling.

```python
import asyncio
from pyrtmp import StreamClosedException
from pyrtmp.messages import SessionManager
from pyrtmp.messages.audio import AudioMessage
from pyrtmp.messages.command import NCConnect, NCCreateStream, NSPublish
from pyrtmp.messages.data import MetaDataMessage
from pyrtmp.messages.protocolcontrol import WindowAcknowledgementSize, SetChunkSize, SetPeerBandwidth
from pyrtmp.messages.usercontrol import StreamBegin
from pyrtmp.messages.video import VideoMessage

async def my_controller(reader, writer):
    # Initialize session manager with reader/writer from asyncio
    session = SessionManager(reader=reader, writer=writer)

    # Access client info
    print(f"Client connected: {session.peername}")
    print(f"Total bytes read: {session.total_read_bytes}")

    try:
        # Perform RTMP handshake (required before chunk processing)
        await session.handshake()

        # Async iterate over incoming chunks
        async for chunk in session.read_chunks_from_stream():
            message = chunk.as_message()
            print(f"Received: {message} chunk_id={message.chunk_id}")

            # Handle connection request
            if isinstance(message, NCConnect):
                session.write_chunk_to_stream(WindowAcknowledgementSize(ack_window_size=5000000))
                session.write_chunk_to_stream(SetPeerBandwidth(ack_window_size=5000000, limit_type=2))
                session.write_chunk_to_stream(StreamBegin(stream_id=0))
                session.write_chunk_to_stream(SetChunkSize(chunk_size=8192))
                session.writer_chunk_size = 8192
                session.write_chunk_to_stream(message.create_response())
                await session.drain()

            elif isinstance(message, SetChunkSize):
                # Update reader chunk size when client sends SetChunkSize
                session.reader_chunk_size = message.chunk_size

            elif isinstance(message, VideoMessage):
                # Process video frame
                print(f"Video timestamp={message.timestamp} size={len(message.payload)}")

            elif isinstance(message, AudioMessage):
                # Process audio frame
                print(f"Audio timestamp={message.timestamp} size={len(message.payload)}")

    except StreamClosedException:
        print(f"Client {session.peername} disconnected")
```

--------------------------------

### RTMPT Send Data Endpoint

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

An endpoint to send data to an existing RTMPT session and receive a response. It feeds data into the session's reader and reads from the buffer.

```python
@app.route('/send/<string:sid>/<int:segment>', methods=['POST'])
async def rtmpt_send(sid: str, segment: int):
    """Send data to RTMPT session and get response"""
    body = await request.body
    sessions[sid].reader.feed_data(body)
    data = await sessions[sid].read_from_buffer()
    resp = quart.Response(data)
    resp.headers['Content-Type'] = 'application/x-fcs'
    return resp
```

--------------------------------

### RTMPT Idle Poll Endpoint

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

An endpoint to poll for data from an RTMPT session without sending any data. It reads data from the session's buffer.

```python
@app.route('/idle/<string:sid>/<int:segment>', methods=['POST'])
async def rtmpt_idle(sid: str, segment: int):
    """Poll for data without sending"""
    data = await sessions[sid].read_from_buffer()
    resp = quart.Response(data)
    resp.headers['Content-Type'] = 'application/x-fcs'
    return resp
```

--------------------------------

### RTMPT Close Session Endpoint

Source: https://context7.com/aungyezawdev/pyrtmp/llms.txt

An endpoint to close an RTMPT session. It reads any remaining data, closes the session, and removes it from the active sessions.

```python
@app.route('/close/<string:sid>/<int:segment>', methods=['POST'])
async def rtmpt_close(sid: str, segment: int):
    """Close RTMPT session"""
    data = await sessions[sid].read_from_buffer()
    sessions[sid].close()
    del sessions[sid]
    return quart.Response(data)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.