SocketChannel

public abstract class SocketChannel
    extends AbstractSelectableChannel
    implements ByteChannel, ScatteringByteChannel, GatheringByteChannel, NetworkChannel {
    
    protected SocketChannel(SelectorProvider provider) {
        super(provider);
    }
    
    // Factory methods
    public static SocketChannel open() throws IOException {
        return SelectorProvider.provider().openSocketChannel();
    }
    
    public static SocketChannel open(ProtocolFamily family) throws IOException {
        return SelectorProvider.provider().openSocketChannel(family);
    }
    
    // Connection management
    public abstract boolean connect(SocketAddress remote) throws IOException;
    public abstract boolean finishConnect() throws IOException;
    public abstract boolean isConnected();
    public abstract boolean isConnectionPending();
    
    // I/O operations
    public abstract int read(ByteBuffer dst) throws IOException;
    public abstract long read(ByteBuffer[] dsts, int offset, int length) throws IOException;
    public abstract int write(ByteBuffer src) throws IOException;
    public abstract long write(ByteBuffer[] srcs, int offset, int length) throws IOException;
    
    // Socket shutdown
    public abstract SocketChannel shutdownInput() throws IOException;
    public abstract SocketChannel shutdownOutput() throws IOException;
    
    // Socket address
    public abstract SocketAddress getRemoteAddress() throws IOException;
    public abstract SocketAddress getLocalAddress() throws IOException;
}

class SocketChannelImpl extends SocketChannel implements SelChImpl {
    private final ProtocolFamily family;
    private final FileDescriptor fd;
    private final int fdVal;
    
    // Three-lock synchronization strategy
    private final ReentrantLock readLock = new ReentrantLock();
    private final ReentrantLock writeLock = new ReentrantLock();
    private final Object stateLock = new Object();
    
    // Connection state machine
    private static final int ST_UNCONNECTED = 0;
    private static final int ST_CONNECTIONPENDING = 1;
    private static final int ST_CONNECTED = 2;
    private static final int ST_CLOSING = 3;
    private static final int ST_CLOSED = 4;
    private volatile int state;
    
    // Socket shutdown state
    private volatile boolean isInputClosed;
    private volatile boolean isOutputClosed;
    private boolean connectionReset;
    
    // Address information
    private SocketAddress localAddress;
    private SocketAddress remoteAddress;
    
    // Virtual thread integration
    private volatile boolean forcedNonBlocking;
    private Thread readerThread;
    private Thread writerThread;
}

// Basic connection (blocking by default)
try (SocketChannel channel = SocketChannel.open()) {
    SocketAddress serverAddress = new InetSocketAddress("example.com", 8080);
    
    // Blocking connect (returns true if connected immediately)
    boolean connected = channel.connect(serverAddress);
    if (!connected && channel.isConnectionPending()) {
        while (!channel.finishConnect()) {
            // Wait for connection completion
            Thread.sleep(10);
        }
    }
    
    // Check connection state
    if (channel.isConnected()) {
        System.out.println("Connected to " + channel.getRemoteAddress());
        System.out.println("Local address: " + channel.getLocalAddress());
    }
}

// Non-blocking connection
try (SocketChannel channel = SocketChannel.open()) {
    channel.configureBlocking(false);
    SocketAddress serverAddress = new InetSocketAddress("example.com", 8080);
    
    boolean connected = channel.connect(serverAddress);
    if (!connected) {
        // Connection in progress, register with selector
        SelectionKey key = channel.register(selector, SelectionKey.OP_CONNECT);
        
        while (true) {
            selector.select();
            if (key.isConnectable()) {
                if (channel.finishConnect()) {
                    key.interestOps(SelectionKey.OP_READ);
                    break;
                }
            }
        }
    }
}

// Internet Protocol (IPv4/IPv6) - default
SocketChannel internetChannel = SocketChannel.open();

// Unix Domain Sockets (inter-process communication on same host)
SocketChannel unixChannel = SocketChannel.open(StandardProtocolFamily.UNIX);
UnixDomainSocketAddress unixAddress = UnixDomainSocketAddress.of("/tmp/mysocket");
unixChannel.connect(unixAddress);

// Protocol family selection based on address type
public static SocketChannel createChannel(SocketAddress address) throws IOException {
    if (address instanceof InetSocketAddress) {
        return SocketChannel.open();
    } else if (address instanceof UnixDomainSocketAddress) {
        return SocketChannel.open(StandardProtocolFamily.UNIX);
    } else {
        throw new IllegalArgumentException("Unsupported address type");
    }
}

// Simple echo client
try (SocketChannel channel = SocketChannel.open()) {
    channel.connect(new InetSocketAddress("localhost", 8080));
    
    // Send message
    String message = "Hello, Server!";
    ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
    while (writeBuffer.hasRemaining()) {
        channel.write(writeBuffer);
    }
    
    // Receive response
    ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    int bytesRead = channel.read(readBuffer);
    if (bytesRead > 0) {
        readBuffer.flip();
        String response = StandardCharsets.UTF_8.decode(readBuffer).toString();
        System.out.println("Server response: " + response);
    }
}

// Handling partial reads/writes
public static void writeFully(SocketChannel channel, ByteBuffer buffer) throws IOException {
    while (buffer.hasRemaining()) {
        int written = channel.write(buffer);
        if (written == 0) {
            // Socket buffer full, wait for write readiness
            if (channel.isBlocking()) {
                // In blocking mode, write will eventually succeed
                continue;
            } else {
                // Non-blocking mode, need to wait for OP_WRITE
                throw new WouldBlockException();
            }
        }
    }
}

public static int readFully(SocketChannel channel, ByteBuffer buffer) throws IOException {
    int totalRead = 0;
    while (buffer.hasRemaining()) {
        int read = channel.read(buffer);
        if (read == -1) {
            return totalRead > 0 ? totalRead : -1; // EOF
        }
        if (read == 0) {
            // No data available
            break;
        }
        totalRead += read;
    }
    return totalRead;
}

// Scattering read - read into multiple buffers
try (SocketChannel channel = SocketChannel.open()) {
    channel.connect(new InetSocketAddress("localhost", 8080));
    
    ByteBuffer header = ByteBuffer.allocate(128);
    ByteBuffer body = ByteBuffer.allocate(1024);
    ByteBuffer[] buffers = {header, body};
    
    // Read into buffers sequentially
    long totalRead = channel.read(buffers);
    System.out.println("Read " + totalRead + " bytes into multiple buffers");
    
    // Process each buffer
    header.flip();
    body.flip();
    processHeader(header);
    processBody(body);
}

// Gathering write - write from multiple buffers
try (SocketChannel channel = SocketChannel.open()) {
    channel.connect(new InetSocketAddress("localhost", 8080));
    
    ByteBuffer header = createHeader();
    ByteBuffer payload = createPayload();
    ByteBuffer[] buffers = {header, payload};
    
    // Write from all buffers
    long totalWritten = channel.write(buffers);
    System.out.println("Wrote " + totalWritten + " bytes from multiple buffers");
}

// Configure socket options before or after connection
try (SocketChannel channel = SocketChannel.open()) {
    // Set send and receive buffer sizes
    channel.setOption(StandardSocketOptions.SO_SNDBUF, 64 * 1024); // 64KB
    channel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024); // 64KB
    
    // Enable TCP_NODELAY (disable Nagle's algorithm)
    channel.setOption(StandardSocketOptions.TCP_NODELAY, true);
    
    // Enable keep-alive
    channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
    
    // Enable address reuse
    channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
    
    // Set linger timeout (seconds to wait on close if data present)
    channel.setOption(StandardSocketOptions.SO_LINGER, 5);
    
    // Set IP Type of Service (TOS) for QoS
    channel.setOption(StandardSocketOptions.IP_TOS, 0x10); // Low delay
    
    // Connect after configuration
    channel.connect(new InetSocketAddress("example.com", 8080));
    
    // Read back option values
    int sendBufferSize = channel.getOption(StandardSocketOptions.SO_SNDBUF);
    boolean tcpNoDelay = channel.getOption(StandardSocketOptions.TCP_NODELAY);
    System.out.println("Send buffer: " + sendBufferSize + ", TCP_NODELAY: " + tcpNoDelay);
}

// Unix domain socket specific options
try (SocketChannel channel = SocketChannel.open(StandardProtocolFamily.UNIX)) {
    // Unix domain sockets support subset of options
    channel.setOption(StandardSocketOptions.SO_SNDBUF, 32 * 1024);
    channel.setOption(StandardSocketOptions.SO_RCVBUF, 32 * 1024);
    channel.setOption(StandardSocketOptions.SO_LINGER, 2);
    
    // Connect to Unix domain socket
    channel.connect(UnixDomainSocketAddress.of("/tmp/mysocket"));
}

// Blocking mode (default)
try (SocketChannel channel = SocketChannel.open()) {
    channel.configureBlocking(true); // Explicitly set blocking
    channel.connect(new InetSocketAddress("example.com", 8080));
    
    // read() and write() will block until data is available/transferred
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    int bytesRead = channel.read(buffer); // Blocks until data arrives or EOF
    
    buffer.flip();
    while (buffer.hasRemaining()) {
        channel.write(buffer); // Blocks until all data written
    }
}

// Non-blocking mode
try (SocketChannel channel = SocketChannel.open()) {
    channel.configureBlocking(false);
    SocketAddress address = new InetSocketAddress("example.com", 8080);
    
    // Non-blocking connect
    boolean connected = channel.connect(address);
    if (!connected) {
        // Use selector to wait for connection completion
        Selector selector = Selector.open();
        channel.register(selector, SelectionKey.OP_CONNECT);
        
        while (true) {
            selector.select();
            if (channel.finishConnect()) {
                break;
            }
        }
    }
    
    // Non-blocking I/O
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    int bytesRead = channel.read(buffer);
    if (bytesRead == 0) {
        // No data available immediately
        // Register for read readiness and continue other work
    } else if (bytesRead > 0) {
        // Data received
        buffer.flip();
        processData(buffer);
    }
}

// With virtual threads, sockets are automatically non-blocking
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> {
        try (SocketChannel channel = SocketChannel.open()) {
            // Even though we don't call configureBlocking(false),
            // the socket will be non-blocking for virtual threads
            channel.connect(new InetSocketAddress("example.com", 8080));
            
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int bytesRead = channel.read(buffer);
            // This read won't block the carrier thread
            // If no data available, virtual thread yields
        }
    });
}

// Manual control of virtual thread behavior
try (SocketChannel channel = SocketChannel.open()) {
    // Check if socket is forced to non-blocking for virtual threads
    Method isForcedNonBlocking = SocketChannelImpl.class.getDeclaredMethod("isForcedNonBlocking");
    isForcedNonBlocking.setAccessible(true);
    boolean forced = (boolean) isForcedNonBlocking.invoke(channel);
    System.out.println("Forced non-blocking: " + forced);
}

try (SocketChannel channel = SocketChannel.open()) {
    channel.connect(new InetSocketAddress("example.com", 8080));
    
    // Perform bidirectional communication
    channel.write(requestBuffer);
    channel.read(responseBuffer);
    
    // Shutdown output (send FIN to peer)
    channel.shutdownOutput();
    System.out.println("Output shutdown, peer will receive EOF on read");
    
    // Continue reading until peer closes connection
    ByteBuffer finalBuffer = ByteBuffer.allocate(1024);
    while (channel.read(finalBuffer) > 0) {
        finalBuffer.flip();
        processFinalData(finalBuffer);
        finalBuffer.clear();
    }
    
    // Shutdown input
    channel.shutdownInput();
    System.out.println("Input shutdown, further reads will return -1");
    
    // Channel can still be closed
    // With SO_LINGER set, close will wait for pending data to be sent
} // Auto-close with try-with-resources

// Half-close pattern for request-response protocols
public static String sendRequestGetResponse(String host, int port, String request) 
        throws IOException {
    try (SocketChannel channel = SocketChannel.open()) {
        channel.connect(new InetSocketAddress(host, port));
        
        // Send request
        ByteBuffer requestBuffer = ByteBuffer.wrap(request.getBytes(StandardCharsets.UTF_8));
        while (requestBuffer.hasRemaining()) {
            channel.write(requestBuffer);
        }
        
        // Half-close: indicate we're done sending
        channel.shutdownOutput();
        
        // Read response until EOF
        ByteBuffer responseBuffer = ByteBuffer.allocate(4096);
        StringBuilder response = new StringBuilder();
        while (channel.read(responseBuffer) != -1) {
            responseBuffer.flip();
            response.append(StandardCharsets.UTF_8.decode(responseBuffer));
            responseBuffer.clear();
        }
        
        return response.toString();
    }
}

public class SimpleTcpClient {
    public static String sendRequest(String host, int port, String request, int timeoutMs) 
            throws IOException {
        try (SocketChannel channel = SocketChannel.open()) {
            // Configure socket
            channel.setOption(StandardSocketOptions.TCP_NODELAY, true);
            channel.setOption(StandardSocketOptions.SO_RCVBUF, 32 * 1024);
            channel.setOption(StandardSocketOptions.SO_SNDBUF, 32 * 1024);
            
            // Connect with timeout
            channel.configureBlocking(true);
            SocketAddress address = new InetSocketAddress(host, port);
            
            // Java doesn't have connect timeout in NIO directly,
            // but we can use async connect with selector
            channel.configureBlocking(false);
            channel.connect(address);
            
            Selector selector = Selector.open();
            channel.register(selector, SelectionKey.OP_CONNECT);
            
            if (selector.select(timeoutMs) == 0) {
                throw new SocketTimeoutException("Connection timed out");
            }
            
            if (!channel.finishConnect()) {
                throw new IOException("Connection failed");
            }
            
            // Send request
            ByteBuffer requestBuffer = ByteBuffer.wrap(request.getBytes(StandardCharsets.UTF_8));
            writeFully(channel, requestBuffer);
            
            // Receive response
            ByteBuffer responseBuffer = ByteBuffer.allocate(4096);
            readFully(channel, responseBuffer);
            responseBuffer.flip();
            
            return StandardCharsets.UTF_8.decode(responseBuffer).toString();
        }
    }
    
    private static void writeFully(SocketChannel channel, ByteBuffer buffer) throws IOException {
        while (buffer.hasRemaining()) {
            channel.write(buffer);
        }
    }
    
    private static void readFully(SocketChannel channel, ByteBuffer buffer) throws IOException {
        while (buffer.hasRemaining()) {
            int read = channel.read(buffer);
            if (read == -1) break;
        }
    }
}

public class NonBlockingTcpClient {
    private final Selector selector;
    private final Map<SocketChannel, ClientSession> sessions;
    
    public NonBlockingTcpClient() throws IOException {
        this.selector = Selector.open();
        this.sessions = new ConcurrentHashMap<>();
    }
    
    public void connect(String host, int port, CompletionHandler<ClientSession> handler) 
            throws IOException {
        SocketChannel channel = SocketChannel.open();
        channel.configureBlocking(false);
        
        SocketAddress address = new InetSocketAddress(host, port);
        boolean connected = channel.connect(address);
        
        ClientSession session = new ClientSession(channel, handler);
        sessions.put(channel, session);
        
        int ops = SelectionKey.OP_CONNECT;
        if (!connected) {
            channel.register(selector, ops, session);
        } else {
            // Connected immediately
            session.onConnected();
            channel.register(selector, SelectionKey.OP_READ, session);
        }
    }
    
    public void processEvents() throws IOException {
        while (true) {
            selector.select();
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iter = selectedKeys.iterator();
            
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                iter.remove();
                
                if (key.isValid()) {
                    if (key.isConnectable()) {
                        handleConnect(key);
                    }
                    if (key.isReadable()) {
                        handleRead(key);
                    }
                    if (key.isWritable()) {
                        handleWrite(key);
                    }
                }
            }
        }
    }
    
    private void handleConnect(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ClientSession session = (ClientSession) key.attachment();
        
        try {
            if (channel.finishConnect()) {
                session.onConnected();
                key.interestOps(SelectionKey.OP_READ);
            }
        } catch (IOException e) {
            session.onConnectionFailed(e);
            sessions.remove(channel);
            channel.close();
        }
    }
    
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ClientSession session = (ClientSession) key.attachment();
        
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int bytesRead = channel.read(buffer);
        
        if (bytesRead == -1) {
            // EOF - connection closed by peer
            session.onDisconnected();
            sessions.remove(channel);
            channel.close();
            key.cancel();
        } else if (bytesRead > 0) {
            buffer.flip();
            session.onDataReceived(buffer);
        }
    }
    
    private static class ClientSession {
        // Session management implementation
    }
}

public class FrameBasedProtocol {
    private static final int HEADER_SIZE = 4; // 4-byte length prefix
    
    public static void sendMessage(SocketChannel channel, ByteBuffer message) throws IOException {
        // Create header with message length
        ByteBuffer header = ByteBuffer.allocate(HEADER_SIZE);
        header.putInt(message.remaining());
        header.flip();
        
        // Gather write: header + message
        ByteBuffer[] buffers = {header, message};
        long totalWritten = 0;
        long expected = HEADER_SIZE + message.remaining();
        
        while (totalWritten < expected) {
            long written = channel.write(buffers);
            if (written == 0) {
                // Would block in non-blocking mode
                break;
            }
            totalWritten += written;
        }
    }
    
    public static ByteBuffer receiveMessage(SocketChannel channel) throws IOException {
        // Scattering read: header first, then body
        ByteBuffer header = ByteBuffer.allocate(HEADER_SIZE);
        int headerRead = 0;
        
        while (header.hasRemaining()) {
            int read = channel.read(header);
            if (read == -1) {
                throw new EOFException("Connection closed while reading header");
            }
            if (read == 0) {
                break; // Would block
            }
            headerRead += read;
        }
        
        if (!header.hasRemaining()) {
            header.flip();
            int messageLength = header.getInt();
            
            if (messageLength > 1024 * 1024) { // 1MB limit
                throw new IOException("Message too large: " + messageLength);
            }
            
            ByteBuffer message = ByteBuffer.allocate(messageLength);
            while (message.hasRemaining()) {
                int read = channel.read(message);
                if (read == -1) {
                    throw new EOFException("Connection closed while reading message");
                }
                if (read == 0) {
                    break; // Would block
                }
            }
            
            if (!message.hasRemaining()) {
                message.flip();
                return message;
            }
        }
        
        return null; // Incomplete message
    }
}

// Simplified lock usage pattern
private int read(ByteBuffer buf) throws IOException {
    readLock.lock();  // Acquire read lock
    try {
        ensureOpenAndConnected();
        // Perform read operation
        return implRead(buf);
    } finally {
        readLock.unlock();
    }
}

private int write(ByteBuffer buf) throws IOException {
    writeLock.lock();  // Acquire write lock
    try {
        ensureOpenAndConnected();
        // Perform write operation
        return implWrite(buf);
    } finally {
        writeLock.unlock();
    }
}

private void updateState(int newState) {
    synchronized (stateLock) {  // Acquire state lock
        // Update connection state
        this.state = newState;
    }
}

// Automatic non-blocking configuration for virtual threads
private void configureSocketNonBlockingIfVirtualThread() {
    if (isBlocking() && Thread.currentThread().isVirtual()) {
        lockedConfigureBlocking(false);
        forcedNonBlocking = true;
    }
}

// Called before all blocking operations
private void beginRead(boolean blocking) {
    configureSocketNonBlockingIfVirtualThread();
    if (blocking && !forcedNonBlocking) {
        begin();  // Setup for interruption
    }
}

// Virtual thread parking during I/O
private void park(int event) {
    if (Thread.currentThread().isVirtual()) {
        Poller.poll(fd, event, -1);  // Park virtual thread
    } else {
        // Platform thread polling
    }
}

// Native method declarations in Net.java
static native int connect(int fd, InetAddress address, int port);
static native int read(int fd, long address, int len) throws IOException;
static native int write(int fd, long address, int len) throws IOException;

// Platform-specific implementations
// Unix: src/java.base/unix/native/libnio/ch/Net.c
// Windows: src/java.base/windows/native/libnio/ch/Net.c

// FileDescriptor management
private static native void initIDs();
static native int socket0(ProtocolFamily family, boolean stream);
static native void bind0(int fd, InetAddress address, int port);

// Optimal buffer sizes based on network characteristics
public static void tuneSocketBuffers(SocketChannel channel, int mtu) throws IOException {
    // For high-bandwidth, high-latency networks (BDP tuning)
    long bandwidth = 100_000_000; // 100 Mbps in bits/sec
    long rtt = 100_000_000; // 100ms in nanoseconds
    long bdp = (bandwidth * rtt) / 8_000_000_000L; // Bytes
    
    int bufferSize = (int) Math.min(bdp, Integer.MAX_VALUE);
    bufferSize = Math.max(bufferSize, 64 * 1024); // At least 64KB
    
    channel.setOption(StandardSocketOptions.SO_SNDBUF, bufferSize);
    channel.setOption(StandardSocketOptions.SO_RCVBUF, bufferSize);
    
    // Enable TCP_NODELAY for interactive applications
    channel.setOption(StandardSocketOptions.TCP_NODELAY, true);
}