Connection Management

class Factory:
    """Socket connection factory for synchronous IRC connections."""
    
    def __init__(self, bind_address=None, wrapper=None, ipv6: bool = False):
        """
        Initialize connection factory.
        
        Parameters:
        - bind_address: tuple, optional local address to bind (host, port)
        - wrapper: callable, socket wrapper function (for SSL, etc.)
        - ipv6: bool, whether to use IPv6 (default: False)
        """
    
    @property
    def family(self) -> int:
        """Socket family (AF_INET or AF_INET6)."""
    
    @property
    def bind_address(self) -> tuple:
        """Local bind address."""
    
    @property
    def wrapper(self) -> callable:
        """Socket wrapper function."""
    
    def connect(self, server_address: tuple):
        """
        Create connection to server.
        
        Parameters:
        - server_address: tuple, (hostname, port)
        
        Returns:
        Connected socket object
        """
    
    def __call__(self, server_address: tuple):
        """
        Callable interface for connection creation.
        
        Parameters:
        - server_address: tuple, (hostname, port)
        
        Returns:
        Connected socket object
        """

class AioFactory:
    """Connection factory for asyncio IRC connections."""
    
    def __init__(self, **kwargs):
        """
        Initialize asyncio connection factory.
        
        Parameters:
        - **kwargs: connection arguments passed to asyncio.create_connection
        """
    
    @property
    def connection_args(self) -> dict:
        """Asyncio connection arguments."""
    
    async def connect(self, protocol_instance, server_address: tuple):
        """
        Create asyncio connection to server.
        
        Parameters:
        - protocol_instance: asyncio protocol instance
        - server_address: tuple, (hostname, port)
        
        Returns:
        (transport, protocol) tuple
        """
    
    async def __call__(self, protocol_instance, server_address: tuple):
        """
        Callable interface for asyncio connection creation.
        
        Parameters:
        - protocol_instance: asyncio protocol instance
        - server_address: tuple, (hostname, port)
        
        Returns:
        (transport, protocol) tuple
        """

def identity(x):
    """
    Identity function for socket wrapper.
    
    Parameters:
    - x: input value
    
    Returns:
    Unchanged input value
    """

class ServerConnection:
    def set_rate_limit(self, frequency: float):
        """
        Set rate limiting for outgoing messages.
        
        Limits the number of messages sent per second to prevent
        server disconnection due to excess flooding.
        
        Parameters:
        - frequency: float, maximum messages per second (e.g., 1.0 = 1 msg/sec)
        """

class AioConnection:
    def set_rate_limit(self, frequency: float):
        """
        Set rate limiting for outgoing messages (asyncio version).
        
        Parameters:
        - frequency: float, maximum messages per second
        """

class ServerConnection:
    def set_keepalive(self, interval: int):
        """
        Set keepalive ping interval.
        
        Automatically sends PING messages to server at regular intervals
        to detect connection issues and prevent timeout disconnections.
        
        Parameters:
        - interval: int, seconds between keepalive pings (0 to disable)
        """

def _ping_ponger(connection, event):
    """
    Default ping response handler.
    
    Automatically responds to server PING messages with appropriate PONG.
    This handler is installed by default on all connections.
    
    Parameters:
    - connection: ServerConnection, IRC connection
    - event: Event, PING event
    """

class ServerConnection:
    @property
    def connected(self) -> bool:
        """Whether connection is established and active."""
    
    @property
    def socket(self):
        """Underlying socket object."""
    
    def get_server_name(self) -> str:
        """
        Get connected server name.
        
        Returns:
        str, server hostname or name
        """
    
    def is_connected(self) -> bool:
        """
        Check if connected to server.
        
        Returns:
        bool, True if connection is active
        """
    
    def reconnect(self):
        """
        Reconnect to the same server using stored parameters.
        
        Attempts to re-establish connection with the same server,
        port, nickname, and other connection parameters.
        """
    
    def close(self):
        """
        Close connection immediately without sending QUIT.
        
        Forcibly closes the socket connection without proper IRC logout.
        Use disconnect() for graceful disconnection.
        """
    
    def disconnect(self, message: str = ""):
        """
        Gracefully disconnect from server.
        
        Sends QUIT message and closes connection properly.
        
        Parameters:
        - message: str, quit message sent to server
        """

import ssl

# SSL context creation
ssl_context = ssl.create_default_context()

# SSL factory for synchronous connections
ssl_factory = Factory(wrapper=ssl_context.wrap_socket)

# SSL factory for asyncio connections  
aio_ssl_factory = AioFactory(ssl=ssl_context)

# IPv6 factory for synchronous connections
ipv6_factory = Factory(ipv6=True)

# IPv6 with SSL
ipv6_ssl_factory = Factory(ipv6=True, wrapper=ssl_context.wrap_socket)

import irc.client
from irc.connection import Factory

# Create custom connection factory
factory = Factory()

# Connect with custom factory
client = irc.client.SimpleIRCClient()
client.connect("irc.libera.chat", 6667, "factorybot", connect_factory=factory)
client.start()

import irc.client
import ssl
from irc.connection import Factory

# Create SSL context
ssl_context = ssl.create_default_context()

# Optional: customize SSL settings
ssl_context.check_hostname = False  # For self-signed certificates
ssl_context.verify_mode = ssl.CERT_NONE  # Disable certificate verification

# Create SSL factory
ssl_factory = Factory(wrapper=ssl_context.wrap_socket)

def on_connect(connection, event):
    print("Connected via SSL!")
    connection.join("#secure")

def on_join(connection, event):
    print(f"Joined {event.target} securely")

client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("join", on_join)

# Connect using SSL on port 6697
client.connect("irc.libera.chat", 6697, "sslbot", connect_factory=ssl_factory)
client.start()

import irc.client
from irc.connection import Factory

# Create IPv6 factory
ipv6_factory = Factory(ipv6=True)

def on_connect(connection, event):
    print("Connected via IPv6!")
    connection.join("#ipv6test")

client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)

# Connect using IPv6
client.connect("irc.libera.chat", 6667, "ipv6bot", connect_factory=ipv6_factory)
client.start()

import irc.client

def on_connect(connection, event):
    print("Connected, configuring rate limiting and keep-alive...")
    
    # Set rate limit to 1 message per second
    connection.set_rate_limit(1.0)
    
    # Send keep-alive ping every 60 seconds
    connection.set_keepalive(60)
    
    connection.join("#ratelimited")

def on_join(connection, event):
    # These messages will be rate-limited
    for i in range(5):
        connection.privmsg(event.target, f"Rate limited message {i+1}")

client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)
client.connection.add_global_handler("join", on_join)

client.connect("irc.libera.chat", 6667, "ratebot")
client.start()

import irc.client
import time

class ReconnectingBot:
    def __init__(self):
        self.client = irc.client.SimpleIRCClient()
        self.server = "irc.libera.chat"
        self.port = 6667
        self.nickname = "reconnectbot"
        self.channels = ["#test"]
        self.setup_handlers()
    
    def setup_handlers(self):
        """Set up event handlers."""
        def on_connect(connection, event):
            print(f"Connected to {connection.get_server_name()}")
            
            # Configure connection
            connection.set_rate_limit(2.0)  # 2 messages per second max
            connection.set_keepalive(30)    # Ping every 30 seconds
            
            # Join channels
            for channel in self.channels:
                connection.join(channel)
        
        def on_disconnect(connection, event):
            print("Disconnected from server")
            self.attempt_reconnection()
        
        def on_ping(connection, event):
            print(f"Received ping from {event.source}")
            # Automatic pong is handled by default ping_ponger
        
        def on_pong(connection, event):
            print(f"Received pong from {event.source}")
        
        def on_pubmsg(connection, event):
            message = event.arguments[0]
            channel = event.target
            
            if message == "!status":
                status = "Connected" if connection.is_connected() else "Disconnected"
                server = connection.get_server_name()
                connection.privmsg(channel, f"Status: {status} to {server}")
            
            elif message == "!reconnect":
                connection.privmsg(channel, "Reconnecting...")
                connection.reconnect()
        
        self.client.connection.add_global_handler("welcome", on_connect)
        self.client.connection.add_global_handler("disconnect", on_disconnect)
        self.client.connection.add_global_handler("ping", on_ping)
        self.client.connection.add_global_handler("pong", on_pong)
        self.client.connection.add_global_handler("pubmsg", on_pubmsg)
    
    def attempt_reconnection(self):
        """Attempt to reconnect with exponential backoff."""
        max_attempts = 5
        base_delay = 2
        
        for attempt in range(max_attempts):
            delay = base_delay * (2 ** attempt)  # Exponential backoff
            print(f"Reconnection attempt {attempt + 1} in {delay} seconds...")
            
            time.sleep(delay)
            
            try:
                self.client.connection.reconnect()
                print("Reconnection successful!")
                return
            except Exception as e:
                print(f"Reconnection failed: {e}")
        
        print("All reconnection attempts failed")
    
    def start(self):
        """Start the bot."""
        self.client.connect(self.server, self.port, self.nickname)
        self.client.start()

# Usage
bot = ReconnectingBot()
bot.start()

import asyncio
import ssl
from irc.client_aio import AioSimpleIRCClient
from irc.connection import AioFactory

async def main():
    # Create SSL context
    ssl_context = ssl.create_default_context()
    
    # Create asyncio SSL factory
    ssl_factory = AioFactory(ssl=ssl_context)
    
    client = AioSimpleIRCClient()
    
    def on_connect(connection, event):
        print("Connected via SSL (asyncio)!")
        connection.join("#asyncssl")
    
    def on_join(connection, event):
        connection.privmsg(event.target, "Hello from asyncio SSL bot!")
    
    client.connection.add_global_handler("welcome", on_connect)
    client.connection.add_global_handler("join", on_join)
    
    # Connect with SSL
    await client.connect(
        "irc.libera.chat", 
        6697, 
        "asyncsslbot", 
        connect_factory=ssl_factory
    )
    
    # Keep running
    await asyncio.Event().wait()

asyncio.run(main())

import irc.client
import socket
from irc.connection import Factory

def logging_wrapper(sock):
    """Custom socket wrapper that logs all data."""
    
    class LoggingSocket:
        def __init__(self, socket):
            self._socket = socket
        
        def send(self, data):
            print(f"SEND: {data}")
            return self._socket.send(data)
        
        def recv(self, size):
            data = self._socket.recv(size)
            print(f"RECV: {data}")
            return data
        
        def __getattr__(self, name):
            return getattr(self._socket, name)
    
    return LoggingSocket(sock)

# Create factory with custom wrapper
logging_factory = Factory(wrapper=logging_wrapper)

def on_connect(connection, event):
    connection.join("#logging")

client = irc.client.SimpleIRCClient()
client.connection.add_global_handler("welcome", on_connect)

# All network traffic will be logged
client.connect("irc.libera.chat", 6667, "loggingbot", connect_factory=logging_factory)
client.start()

import irc.client
from irc.connection import Factory
import ssl

class MultiServerBot:
    def __init__(self):
        self.reactor = irc.client.Reactor()
        self.connections = {}
        self.setup_servers()
    
    def setup_servers(self):
        """Set up connections to multiple servers."""
        servers = [
            {
                "name": "libera",
                "host": "irc.libera.chat",
                "port": 6697,
                "ssl": True,
                "channels": ["#test1", "#libera"]
            },
            {
                "name": "oftc", 
                "host": "irc.oftc.net",
                "port": 6667,
                "ssl": False,
                "channels": ["#test2", "#oftc"]
            }
        ]
        
        for server_config in servers:
            self.setup_server_connection(server_config)
    
    def setup_server_connection(self, config):
        """Set up connection to a single server."""
        connection = self.reactor.server()
        
        # Create appropriate factory
        if config["ssl"]:
            ssl_context = ssl.create_default_context()
            factory = Factory(wrapper=ssl_context.wrap_socket)
        else:
            factory = Factory()
        
        def on_connect(conn, event):
            print(f"Connected to {config['name']}")
            conn.set_rate_limit(1.0)
            conn.set_keepalive(60)
            
            for channel in config["channels"]:
                conn.join(channel)
        
        def on_pubmsg(conn, event):
            message = event.arguments[0]
            channel = event.target
            nick = event.source.nick
            
            print(f"[{config['name']}:{channel}] <{nick}> {message}")
            
            if message == "!servers":
                server_list = list(self.connections.keys())
                conn.privmsg(channel, f"Connected to: {', '.join(server_list)}")
        
        connection.add_global_handler("welcome", on_connect)
        connection.add_global_handler("pubmsg", on_pubmsg)
        
        # Connect to server
        connection.connect(
            config["host"],
            config["port"], 
            f"multibot_{config['name']}",
            connect_factory=factory
        )
        
        self.connections[config["name"]] = connection
    
    def start(self):
        """Start processing events for all connections."""
        print("Starting multi-server bot...")
        self.reactor.process_forever()

# Usage
bot = MultiServerBot()
bot.start()