aioice

Warnings

• breaking aioice versions 0.9.0 and higher require Python 3.9 or newer. Users on older Python versions must upgrade their Python environment or use an older, unsupported version of aioice.
  Fix: Upgrade Python to 3.9 or a later supported version.
• gotcha When working with `asyncio`-based libraries like aioice, ensure you use `await asyncio.sleep(duration)` instead of `time.sleep(duration)`. Using `time.sleep` will block the entire asyncio event loop, causing your application to freeze and potentially leading to connection timeouts or failures.
  Fix: Replace `time.sleep()` with `await asyncio.sleep()` in all asynchronous contexts.
• gotcha While aioice supports 'half-trickle ICE' (adding candidates iteratively), a full understanding of ICE trickle mechanisms is important. Ensure your signaling protocol properly handles the exchange of candidates over time, and that both peers support the desired trickle behavior to avoid delays or connection failures, especially with older client implementations.
  Fix: Review ICE trickle specifications (RFC 8838) and ensure your signaling mechanism and peer implementations are compatible with the expected candidate exchange patterns. Explicitly signal end-of-candidates with `connection.add_remote_candidate(None)`.
• gotcha The error 'STUN transaction failed (400 - You cannot use the same channel number with different peer)' can occur due to race conditions during channel binding or incorrect handling of channel numbers, particularly when rapidly creating or re-establishing connections. This can lead to intermittent connection issues.
  Fix: Implement robust state management for ICE connections and channel binding. Ensure proper synchronization and unique channel identifier allocation for each peer and component to prevent conflicts.

Install

• pip install aioice Install latest version

Imports

• Connection

  from aioice import Connection

• Candidate

  from aioice import Candidate

Quickstart

This quickstart demonstrates the basic steps to establish an ICE connection: creating a `Connection` object, gathering local candidates, exchanging candidate information (and username/password) with a remote peer via a signaling channel, adding remote candidates, performing the ICE handshake, and finally sending/receiving data. Note that the signaling part is a placeholder for your specific application's communication method.

import asyncio
import aioice
import os

async def connect_using_ice():
    # In a real application, STUN/TURN servers should be provided.
    # For testing, you might use public STUN servers or set up your own.
    stun_server = os.environ.get('AIOICE_STUN_SERVER', 'stun.l.google.com:19302')
    
    connection = aioice.Connection(ice_controlling=True, stun_server=(stun_server.split(':')[0], int(stun_server.split(':')[1])))
    
    # Gather local candidates
    await connection.gather_candidates()
    
    # In a real application, 'send_local_info' and 'get_remote_info'
    # would be replaced by your signaling mechanism (e.g., WebSockets).
    # For this example, we'll simulate an exchange.
    
    # These would be exchanged with the remote peer via signaling
    local_candidates_sdp = [c.to_sdp() for c in connection.local_candidates]
    local_username = connection.local_username
    local_password = connection.local_password
    
    print(f"Local candidates: {local_candidates_sdp}")
    print(f"Local username: {local_username}")
    print(f"Local password: {local_password}")
    
    # --- Simulate remote information reception (replace with actual signaling) ---
    # For a full example, you'd receive this from another peer.
    # Let's assume remote_candidates, remote_username, remote_password are obtained.
    remote_candidates = [] # Populate with remote Candidate objects or SDP strings
    remote_username = "remote_user" # Example
    remote_password = "remote_pass" # Example
    # --- End simulation ---

    # Add remote candidates
    for candidate_sdp in remote_candidates:
        await connection.add_remote_candidate(aioice.Candidate.from_sdp(candidate_sdp))
    await connection.add_remote_candidate(None) # Signal end-of-candidates

    connection.remote_username = remote_username
    connection.remote_password = remote_password
    
    print("Performing ICE handshake...")
    try:
        await connection.connect()
        print(f"ICE connection established: {connection.state}")

        # Send and receive data on component 1
        await connection.sendto(b'Hello from aioice!', 1)
        print("Sent 'Hello from aioice!'")
        
        # In a real app, you would continuously listen for data
        data, component = await connection.recvfrom()
        print(f"Received '{data.decode()}' on component {component}")

    except Exception as e:
        print(f"ICE connection failed: {e}")
    finally:
        await connection.close()
        print("Connection closed.")

# To run this, you'd typically have two peers exchanging information via a signaling channel.
# For a local test, one would act as controlling, the other as controlled, and manually
# exchange the candidate, username, and password strings.
# asyncio.run(connect_using_ice())
print("Quickstart example demonstrates aioice usage. For a full connection, actual signaling is required.")