Core I/O Abstractions

/**
 * Core abstraction for I/O communication using bytes (non-blocking)
 */
interface EndPoint extends Closeable {
    /** Read data into buffer, returns number of bytes read or -1 for EOF */
    int fill(ByteBuffer buffer) throws IOException;
    
    /** Write data from buffers, returns true if all data was written */
    boolean flush(ByteBuffer... buffers) throws IOException;
    
    /** Register callback for async read notification when data is available */
    void fillInterested(Callback callback) throws ReadPendingException;
    
    /** Async write operation with callback notification */
    void write(Callback callback, ByteBuffer... buffers) throws WritePendingException;
    
    /** Connectionless read operation for UDP-style protocols */
    SocketAddress receive(ByteBuffer buffer) throws IOException;
    
    /** Connectionless write operation for UDP-style protocols */
    boolean send(SocketAddress address, ByteBuffer... buffers) throws IOException;
    
    /** Check if endpoint is open for I/O operations */
    boolean isOpen();
    
    /** Close endpoint with optional cause */
    void close(Throwable cause);
    
    /** Shutdown output direction of endpoint */
    void shutdownOutput();
    
    /** Check if output is shutdown */
    boolean isOutputShutdown();
    
    /** Check if input is shutdown */
    boolean isInputShutdown();
    
    /** Get idle timeout in milliseconds */
    long getIdleTimeout();
    
    /** Set idle timeout in milliseconds */
    void setIdleTimeout(long idleTimeout);
    
    /** Get associated connection */
    Connection getConnection();
    
    /** Set associated connection */
    void setConnection(Connection connection);
    
    /** Upgrade to new connection, replacing current one */
    void upgrade(Connection newConnection);
    
    /** Get local socket address */
    SocketAddress getLocalSocketAddress();
    
    /** Get remote socket address */
    SocketAddress getRemoteSocketAddress();
    
    /** Get creation timestamp */
    long getCreatedTimeStamp();
    
    /** Check if connection is secure (SSL/TLS) */
    boolean isSecure();
    
    /** Get SSL session data if available */
    default SslSessionData getSslSessionData() {
        return null;
    }
    
    /** Get underlying transport object */
    Object getTransport();
    
    /** Try to set fill interest without throwing exception */
    boolean tryFillInterested(Callback callback);
    
    /** Check if fill interested is currently set */
    boolean isFillInterested();
    
    /** Callback for when endpoint is opened */
    void onOpen();
    
    /** Callback for when endpoint is closed */
    void onClose(Throwable cause);
    
    // Constants
    SocketAddress EOF = InetSocketAddress.createUnresolved("", 0);
    
    interface SslSessionData {
        SSLSession sslSession();
        String sslSessionId();
        String cipherSuite();
        X509Certificate[] peerCertificates();
        default int keySize() {
            String cipherSuite = cipherSuite();
            return cipherSuite == null ? 0 : SslContextFactory.deduceKeyLength(cipherSuite);
        }
    }
    
    interface Wrapper {
        EndPoint unwrap();
    }
    
    interface Pipe {
        EndPoint getLocalEndPoint();
        EndPoint getRemoteEndPoint();
    }
}

// Basic I/O operations
EndPoint endpoint = // ... obtain endpoint
ByteBuffer buffer = ByteBuffer.allocate(1024);

// Blocking read
int bytesRead = endpoint.fill(buffer);
if (bytesRead > 0) {
    buffer.flip();
    // Process data
}

// Non-blocking write with callback
ByteBuffer data = ByteBuffer.wrap("Hello".getBytes());
endpoint.write(new Callback() {
    @Override
    public void succeeded() {
        System.out.println("Write completed");
    }
    
    @Override
    public void failed(Throwable x) {
        System.err.println("Write failed: " + x.getMessage());
    }
}, data);

// Async read notification
endpoint.fillInterested(new Callback() {
    @Override
    public void succeeded() {
        // Data is now available for reading
        try {
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int bytes = endpoint.fill(buffer);
            // Process data
        } catch (IOException e) {
            failed(e);
        }
    }
    
    @Override
    public void failed(Throwable x) {
        System.err.println("Read failed: " + x.getMessage());
    }
});

/**
 * Connection associated with an EndPoint for processing I/O events
 */
interface Connection extends Closeable {
    /** Called when connection is opened */
    void onOpen();
    
    /** Called when connection is closed */
    void onClose(Throwable cause);
    
    /** Get associated endpoint */
    EndPoint getEndPoint();
    
    /** Handle idle timeout expiration, return true if connection should be closed */
    boolean onIdleExpired(TimeoutException timeoutException);
    
    /** Get number of messages received */
    long getMessagesIn();
    
    /** Get number of messages sent */
    long getMessagesOut();
    
    /** Get number of bytes received */
    long getBytesIn();
    
    /** Get number of bytes sent */
    long getBytesOut();
    
    /** Get connection creation timestamp */
    long getCreatedTimeStamp();
    
    /** Add event listener */
    void addEventListener(EventListener listener);
    
    /** Remove event listener */
    void removeEventListener(EventListener listener);
    
    interface UpgradeFrom {
        /** Produce unconsumed buffer during protocol upgrade */
        ByteBuffer onUpgradeFrom();
    }
    
    interface UpgradeTo {
        /** Receive unconsumed buffer during protocol upgrade */
        void onUpgradeTo(ByteBuffer prefilled);
    }
    
    interface Listener extends EventListener {
        default void onOpened(Connection connection) {}
        default void onClosed(Connection connection) {}
    }
}

public class EchoConnection extends AbstractConnection {
    public EchoConnection(EndPoint endPoint, Executor executor) {
        super(endPoint, executor);
    }
    
    @Override
    public void onOpen() {
        super.onOpen();
        fillInterested(); // Start reading
    }
    
    @Override
    public void onFillable() {
        try {
            ByteBuffer buffer = getByteBufferPool().acquire(1024, false).getByteBuffer();
            try {
                int filled = getEndPoint().fill(buffer);
                if (filled > 0) {
                    buffer.flip();
                    // Echo the data back
                    getEndPoint().write(Callback.NOOP, buffer);
                } else if (filled < 0) {
                    getEndPoint().close();
                }
            } finally {
                getByteBufferPool().release(buffer);
            }
        } catch (IOException e) {
            close(e);
        }
    }
}

abstract class AbstractEndPoint extends AbstractLifeCycle implements EndPoint {
    protected AbstractEndPoint(Scheduler scheduler);
    
    // Common endpoint functionality implemented
    public void close(Throwable cause);
    public boolean isOpen();
    public long getCreatedTimeStamp();
    public void setIdleTimeout(long idleTimeout);
    public long getIdleTimeout();
    
    // Template methods for subclasses
    protected abstract void doClose();
    protected abstract boolean needsFillInterest();
    protected abstract void onIncompleteFlush();
}

abstract class AbstractConnection extends AbstractLifeCycle implements Connection {
    protected AbstractConnection(EndPoint endPoint, Executor executor);
    
    // Common connection functionality
    public EndPoint getEndPoint();
    public long getCreatedTimeStamp();
    public long getMessagesIn();
    public long getMessagesOut();
    public long getBytesIn();
    public long getBytesOut();
    
    // Utility methods for subclasses
    protected void fillInterested();
    protected void fillInterested(Callback callback);
    protected ByteBufferPool getByteBufferPool();
    protected Executor getExecutor();
    
    // Template method for handling readable events
    public abstract void onFillable();
}

interface Connection.Listener extends EventListener {
    /** Called when connection is opened */
    default void onOpened(Connection connection) {}
    
    /** Called when connection is closed */
    default void onClosed(Connection connection) {}
}

interface Connection.UpgradeFrom {
    /** 
     * Produce any unconsumed input during protocol upgrade
     * @return ByteBuffer containing unconsumed data, or null
     */
    ByteBuffer onUpgradeFrom();
}

interface Connection.UpgradeTo {
    /**
     * Receive any unconsumed input during protocol upgrade
     * @param prefilled ByteBuffer containing unconsumed data from previous protocol
     */
    void onUpgradeTo(ByteBuffer prefilled);
}

// HTTP/1.1 to HTTP/2 upgrade
public class HTTP11Connection extends AbstractConnection implements Connection.UpgradeFrom {
    @Override
    public ByteBuffer onUpgradeFrom() {
        // Return any unprocessed HTTP/1.1 data
        return unprocessedBuffer;
    }
}

public class HTTP2Connection extends AbstractConnection implements Connection.UpgradeTo {
    @Override
    public void onUpgradeTo(ByteBuffer prefilled) {
        // Process any data from previous HTTP/1.1 connection
        if (prefilled != null && prefilled.hasRemaining()) {
            processPrefilledData(prefilled);
        }
    }
}

// Perform upgrade
Connection newConnection = new HTTP2Connection(endPoint, executor);
endPoint.upgrade(newConnection);