aioice

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.")