Connection Management

/**
 * Factory for client-side Connection instances
 */
interface ClientConnectionFactory {
    /** 
     * Create a new connection for the given endpoint and context
     * @param endPoint the endpoint for the connection
     * @param context connection context containing configuration and state
     * @return new Connection instance
     * @throws IOException if connection creation fails
     */
    Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException;
    
    /** 
     * Customize a connection after creation
     * @param connection the connection to customize
     * @param context connection context
     * @return the customized connection (may be the same instance or a wrapper)
     */
    default Connection customize(Connection connection, Map<String, Object> context) {
        return connection;
    }
    
    // Context key for client information
    String CLIENT_CONTEXT_KEY = "client";
    
    interface Decorator {
        /** Decorate (wrap) a connection factory */
        ClientConnectionFactory decorate(ClientConnectionFactory factory);
    }
    
    abstract class Info {
        private final String protocol;
        private final ClientConnectionFactory factory;
        
        protected Info(String protocol, ClientConnectionFactory factory);
        
        public String getProtocol();
        public ClientConnectionFactory getClientConnectionFactory();
        public boolean matches(String protocol, String protocols);
    }
}

// HTTP/1.1 client connection factory
ClientConnectionFactory http11Factory = new ClientConnectionFactory() {
    @Override
    public Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException {
        HttpClient client = (HttpClient) context.get(CLIENT_CONTEXT_KEY);
        return new HttpConnection(endPoint, client);
    }
};

// WebSocket client connection factory  
ClientConnectionFactory wsFactory = new ClientConnectionFactory() {
    @Override
    public Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException {
        WebSocketClient client = (WebSocketClient) context.get(CLIENT_CONTEXT_KEY);
        return new WebSocketConnection(endPoint, client);
    }
};

// Decorating factory for logging
ClientConnectionFactory.Decorator loggingDecorator = factory -> new ClientConnectionFactory() {
    @Override
    public Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException {
        System.out.println("Creating connection to: " + endPoint.getRemoteSocketAddress());
        Connection connection = factory.newConnection(endPoint, context);
        return new LoggingConnection(connection);
    }
};

ClientConnectionFactory decoratedFactory = loggingDecorator.decorate(http11Factory);

// Using connection context
Map<String, Object> context = new HashMap<>();
context.put(ClientConnectionFactory.CLIENT_CONTEXT_KEY, httpClient);
context.put("request.timeout", Duration.ofSeconds(30));

EndPoint endPoint = // ... create endpoint
Connection connection = http11Factory.newConnection(endPoint, context);

/**
 * Low-level transport abstraction (TCP, UDP, Unix sockets, etc.)
 */
interface Transport {
    /** Check if transport provides intrinsic security (e.g., Unix domain sockets) */
    boolean isIntrinsicallySecure();
    
    /** Check if transport requires domain name resolution */
    boolean requiresDomainNameResolution();
    
    /** 
     * Connect to remote address
     * @param socketAddress target address to connect to
     * @param context connection context
     * @throws IOException if connection fails
     */
    void connect(SocketAddress socketAddress, Map<String, Object> context) throws IOException;
    
    /** Get the socket address for this transport */
    SocketAddress getSocketAddress();
    
    /** Create a new selectable channel for this transport */
    SelectableChannel newSelectableChannel() throws IOException;
    
    /** 
     * Create new endpoint for this transport
     * @param scheduler scheduler for endpoint operations
     * @param selector managed selector for NIO operations  
     * @param channel the selectable channel
     * @param key the selection key
     * @return new EndPoint instance
     */
    EndPoint newEndPoint(Scheduler scheduler, ManagedSelector selector, 
                        SelectableChannel channel, SelectionKey key);
    
    /** 
     * Create new connection for this transport
     * @param endPoint the endpoint
     * @param context connection context
     * @return new Connection instance
     * @throws IOException if connection creation fails
     */
    Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException;
    
    // Built-in transport instances
    Transport TCP_IP = new TCPIP();
    Transport UDP_IP = new UDPIP();
    
    abstract class Socket implements Transport {
        public abstract SocketAddress getSocketAddress();
        public boolean isIntrinsicallySecure();
        public boolean requiresDomainNameResolution();
    }
    
    abstract class IP extends Socket {
        protected final String host;
        protected final int port;
        
        protected IP(String host, int port);
        
        public String getHost();
        public int getPort();
        public SocketAddress getSocketAddress();
        public boolean requiresDomainNameResolution();
    }
    
    class TCPIP extends IP {
        public TCPIP(String host, int port);
        public SelectableChannel newSelectableChannel() throws IOException;
        public EndPoint newEndPoint(Scheduler scheduler, ManagedSelector selector, 
                                   SelectableChannel channel, SelectionKey key);
        public void connect(SocketAddress address, Map<String, Object> context) throws IOException;
    }
    
    class UDPIP extends IP {
        public UDPIP(String host, int port);
        public SelectableChannel newSelectableChannel() throws IOException;
        public EndPoint newEndPoint(Scheduler scheduler, ManagedSelector selector,
                                   SelectableChannel channel, SelectionKey key);
        public void connect(SocketAddress address, Map<String, Object> context) throws IOException;
    }
    
    abstract class Unix extends Socket {
        protected final Path path;
        
        protected Unix(Path path);
        
        public Path getPath();
        public SocketAddress getSocketAddress();
        public boolean isIntrinsicallySecure();
        public boolean requiresDomainNameResolution();
    }
    
    class TCPUnix extends Unix {
        public TCPUnix(Path path);
        public SelectableChannel newSelectableChannel() throws IOException;
        public EndPoint newEndPoint(Scheduler scheduler, ManagedSelector selector,
                                   SelectableChannel channel, SelectionKey key);
    }
    
    class UDPUnix extends Unix {
        public UDPUnix(Path path);
        public SelectableChannel newSelectableChannel() throws IOException;
        public EndPoint newEndPoint(Scheduler scheduler, ManagedSelector selector,
                                   SelectableChannel channel, SelectionKey key);
    }
    
    class Wrapper implements Transport {
        private final Transport wrapped;
        
        public Wrapper(Transport wrapped);
        public Transport getWrapped();
        
        // All methods delegate to wrapped transport
        public boolean isIntrinsicallySecure();
        public boolean requiresDomainNameResolution();
        // ... other delegating methods
    }
}

// TCP/IP transport
Transport tcpTransport = new Transport.TCPIP("example.com", 8080);
System.out.println("Requires DNS resolution: " + tcpTransport.requiresDomainNameResolution());
System.out.println("Socket address: " + tcpTransport.getSocketAddress());

// UDP/IP transport
Transport udpTransport = new Transport.UDPIP("224.0.0.1", 5000); // Multicast
SelectableChannel udpChannel = udpTransport.newSelectableChannel();

// Unix domain socket transport
Path socketPath = Paths.get("/tmp/app.sock");
Transport unixTransport = new Transport.TCPUnix(socketPath);
System.out.println("Intrinsically secure: " + unixTransport.isIntrinsicallySecure());

// Transport wrapper for custom behavior
Transport loggingTransport = new Transport.Wrapper(tcpTransport) {
    @Override
    public void connect(SocketAddress address, Map<String, Object> context) throws IOException {
        System.out.println("Connecting to: " + address);
        super.connect(address, context);
        System.out.println("Connected successfully");
    }
};

// Creating endpoint with transport
Scheduler scheduler = // ... scheduler instance
ManagedSelector selector = // ... managed selector instance
SelectableChannel channel = tcpTransport.newSelectableChannel();
SelectionKey key = channel.register(selector.getSelector(), SelectionKey.OP_CONNECT);
EndPoint endPoint = tcpTransport.newEndPoint(scheduler, selector, channel, key);

// Creating connection with transport
Map<String, Object> context = new HashMap<>();
Connection connection = tcpTransport.newConnection(endPoint, context);

/**
 * Client-side connector for establishing connections
 */
class ClientConnector extends ContainerLifeCycle {
    public ClientConnector();
    
    // Configuration
    public void setSelectors(int selectors);
    public int getSelectors();
    
    public void setIdleTimeout(Duration idleTimeout);
    public Duration getIdleTimeout();
    
    public void setConnectTimeout(Duration connectTimeout);
    public Duration getConnectTimeout();
    
    public void setByteBufferPool(ByteBufferPool pool);
    public ByteBufferPool getByteBufferPool();
    
    public void setScheduler(Scheduler scheduler);
    public Scheduler getScheduler();
    
    public void setExecutor(Executor executor);
    public Executor getExecutor();
    
    public void setSslContextFactory(SslContextFactory.Client sslContextFactory);
    public SslContextFactory.Client getSslContextFactory();
    
    // Connection establishment
    public CompletableFuture<Connection> connect(SocketAddress address, Map<String, Object> context);
    public CompletableFuture<Connection> connect(Transport transport, Map<String, Object> context);
    
    // Statistics
    public ConnectionStatistics getConnectionStatistics();
}

// Basic client connector setup
ClientConnector connector = new ClientConnector();
connector.setSelectors(4); // 4 selector threads
connector.setIdleTimeout(Duration.ofSeconds(30));
connector.setConnectTimeout(Duration.ofSeconds(15));

// Custom buffer pool
ByteBufferPool customPool = new ArrayByteBufferPool(256, 2, 65536, 64);
connector.setByteBufferPool(customPool);

// Custom scheduler and executor
connector.setScheduler(new ScheduledExecutorScheduler("client-scheduler", false));
connector.setExecutor(new QueuedThreadPool("client-executor"));

// SSL configuration
SslContextFactory.Client sslContextFactory = new SslContextFactory.Client();
sslContextFactory.setTrustAll(false);
sslContextFactory.setEndpointIdentificationAlgorithm("HTTPS");
connector.setSslContextFactory(sslContextFactory);

// Start connector
connector.start();

// Connect to server
SocketAddress address = new InetSocketAddress("example.com", 443);
Map<String, Object> context = new HashMap<>();
context.put(ClientConnectionFactory.CLIENT_CONTEXT_KEY, httpClient);

CompletableFuture<Connection> connectionFuture = connector.connect(address, context);
connectionFuture.thenAccept(connection -> {
    System.out.println("Connected to: " + connection.getEndPoint().getRemoteSocketAddress());
    // Use connection
}).exceptionally(throwable -> {
    System.err.println("Connection failed: " + throwable.getMessage());
    return null;
});

// Connect with custom transport
Transport customTransport = new Transport.TCPIP("example.com", 8080);
CompletableFuture<Connection> customConnection = connector.connect(customTransport, context);

// Monitor connection statistics
ConnectionStatistics stats = connector.getConnectionStatistics();
System.out.println("Total connections: " + stats.getConnectionsTotal());
System.out.println("Open connections: " + stats.getConnectionsOpened());

/**
 * Client connection that can negotiate protocols
 */
class NegotiatingClientConnection extends AbstractConnection {
    protected NegotiatingClientConnection(EndPoint endPoint, Executor executor, 
                                        ClientConnectionFactory connectionFactory,
                                        Map<String, Object> context);
    
    // Protocol negotiation
    protected void negotiate();
    protected String selectProtocol(List<String> protocols);
    protected Connection newConnection(String protocol);
    
    // Lifecycle
    public void onOpen();
    public void onFillable();
    public boolean onIdleExpired(TimeoutException timeoutException);
}

/**
 * Factory for negotiating client connections
 */
class NegotiatingClientConnectionFactory implements ClientConnectionFactory {
    public NegotiatingClientConnectionFactory(ClientConnectionFactory connectionFactory,
                                            String... protocols);
    
    public Connection newConnection(EndPoint endPoint, Map<String, Object> context) throws IOException;
    
    public List<String> getProtocols();
    public ClientConnectionFactory getClientConnectionFactory();
}

// HTTP/2 with HTTP/1.1 fallback
ClientConnectionFactory http2Factory = new HTTP2ClientConnectionFactory();
ClientConnectionFactory http11Factory = new HTTP11ClientConnectionFactory();

NegotiatingClientConnectionFactory negotiatingFactory = 
    new NegotiatingClientConnectionFactory(http2Factory, "h2", "http/1.1");

// ALPN negotiation will occur during SSL handshake
Map<String, Object> context = new HashMap<>();
context.put(ClientConnectionFactory.CLIENT_CONTEXT_KEY, httpClient);

Connection connection = negotiatingFactory.newConnection(sslEndPoint, context);
// Connection will negotiate to HTTP/2 if supported, fallback to HTTP/1.1

// Custom negotiation logic
NegotiatingClientConnection customNegotiation = new NegotiatingClientConnection(
    endPoint, executor, connectionFactory, context) {
    
    @Override
    protected String selectProtocol(List<String> protocols) {
        // Custom protocol selection logic
        if (protocols.contains("custom-protocol-v2")) {
            return "custom-protocol-v2";
        } else if (protocols.contains("custom-protocol-v1")) {
            return "custom-protocol-v1";
        }
        return super.selectProtocol(protocols);
    }
    
    @Override
    protected Connection newConnection(String protocol) {
        switch (protocol) {
            case "custom-protocol-v2":
                return new CustomV2Connection(getEndPoint(), getExecutor());
            case "custom-protocol-v1":
                return new CustomV1Connection(getEndPoint(), getExecutor());
            default:
                return super.newConnection(protocol);
        }
    }
};

/**
 * Statistics collection for connections
 */
class ConnectionStatistics extends AbstractLifeCycle implements Connection.Listener {
    public ConnectionStatistics();
    
    // Connection counts
    public long getConnectionsTotal();
    public long getConnectionsOpened();
    public long getConnectionsClosed();
    public long getConnectionsMax();
    
    // Message counts
    public long getMessagesIn();
    public long getMessagesOut();
    
    // Byte counts  
    public long getBytesIn();
    public long getBytesOut();
    
    // Timing statistics
    public long getConnectionDurationMax();
    public double getConnectionDurationMean();
    public double getConnectionDurationStdDev();
    
    // Rate statistics
    public double getConnectionsPerSecond();
    public double getMessagesInPerSecond();
    public double getMessagesOutPerSecond();
    
    // Reset statistics
    public void reset();
    
    // Connection listener methods
    public void onOpened(Connection connection);
    public void onClosed(Connection connection);
}

// Monitor connection statistics
ConnectionStatistics stats = new ConnectionStatistics();

// Add as listener to connections
connection.addEventListener(stats);

// Or add to connector for all connections
ClientConnector connector = new ClientConnector();
connector.addBean(stats);

// Monitor statistics
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
    @Override
    public void run() {
        System.out.printf("Connections: total=%d, open=%d, closed=%d%n",
            stats.getConnectionsTotal(),
            stats.getConnectionsOpened(), 
            stats.getConnectionsClosed());
        System.out.printf("Throughput: %.2f conn/sec, %.2f KB/sec in, %.2f KB/sec out%n",
            stats.getConnectionsPerSecond(),
            stats.getBytesIn() / 1024.0,
            stats.getBytesOut() / 1024.0);
    }
}, 0, 5000); // Every 5 seconds

// Include/exclude based connection statistics
IncludeExcludeConnectionStatistics filteredStats = new IncludeExcludeConnectionStatistics();
filteredStats.include("https://api.example.com:*");
filteredStats.exclude("https://internal.example.com:*");

// Only connections matching include patterns and not matching exclude patterns
// will be counted in statistics