/home/leaf/crossfire/jxclient/trunk/network/src/main/java/com/realtime/crossfire/jxclient/server/socket/ClientSocket.java

Go to the documentation of this file.

/*
 * This file is part of JXClient, the Fullscreen Java Crossfire Client.
 *
 * JXClient is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * JXClient is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with JXClient; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 * Copyright (C) 2005-2008 Yann Chachkoff.
 * Copyright (C) 2006-2011 Andreas Kirschbaum.
 */

package com.realtime.crossfire.jxclient.server.socket;

import com.realtime.crossfire.jxclient.server.crossfire.Model;
import com.realtime.crossfire.jxclient.util.DebugWriter;
import com.realtime.crossfire.jxclient.util.EventListenerList2;
import java.io.EOFException;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.UnresolvedAddressException;
import java.util.Collection;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

public class ClientSocket {

    private static final int MAXIMUM_PACKET_SIZE = 65536;

    @NotNull
    private final Model model;

    @Nullable
    private final DebugWriter debugProtocol;

    @NotNull
    private final EventListenerList2<ClientSocketListener> clientSocketListeners = new EventListenerList2<ClientSocketListener>(ClientSocketListener.class);

    @NotNull
    private final Selector selector;

    @NotNull
    private final Object syncConnect = new Object();

    private boolean reconnect = true;

    @NotNull
    private String reconnectReason = "disconnect";

    private boolean reconnectIsError = false;

    @Nullable
    private String host = null;

    private int port = 0;

    private boolean disconnectPending = false;

    @NotNull
    private final byte[] packetHeader = new byte[2];

    @Nullable
    private SelectableChannel selectableChannel = null;

    @Nullable
    private SelectionKey selectionKey = null;

    private int interestOps = 0;

    @NotNull
    private final byte[] inputBuf = new byte[2+MAXIMUM_PACKET_SIZE];

    @NotNull
    private final ByteBuffer inputBuffer = ByteBuffer.wrap(inputBuf);

    private int inputLen = -1;

    @NotNull
    private final Object syncOutput = new Object();

    @NotNull
    private final ByteBuffer outputBuffer = ByteBuffer.allocate(2+MAXIMUM_PACKET_SIZE);

    @Nullable
    private SocketChannel socketChannel = null;

    private boolean isConnected = false;

    @NotNull
    private final Thread thread = new Thread(new Runnable() {

        @Override
        public void run() {
            process();
        }

    }, "JXClient:ClientSocket");

    public ClientSocket(@NotNull final Model model, @Nullable final DebugWriter debugProtocol) throws IOException {
        this.model = model;
        this.debugProtocol = debugProtocol;
        selector = Selector.open();
    }

    public void start() {
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:start");
        }
        thread.start();
    }

    public void stop() throws InterruptedException {
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:stop");
        }
        thread.interrupt();
        thread.join();
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:stopped");
        }
    }

    public void addClientSocketListener(@NotNull final ClientSocketListener clientSocketListener) {
        clientSocketListeners.add(clientSocketListener);
    }

    public void removeClientSocketListener(@NotNull final ClientSocketListener clientSocketListener) {
        clientSocketListeners.remove(clientSocketListener);
    }

    public void connect(@NotNull final String host, final int port) {
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:connect "+host+":"+port);
        }
        synchronized (syncConnect) {
            if (this.host == null || this.port == 0 || !this.host.equals(host) || this.port != port) {
                reconnect = true;
                reconnectReason = "connect";
                reconnectIsError = false;
                this.host = host;
                this.port = port;
                selector.wakeup();
            }
        }
    }

    public void disconnect(@NotNull final String reason, final boolean isError) {
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:disconnect: "+reason+(isError ? " [unexpected]" : ""));
        }
        synchronized (syncConnect) {
            if (host != null || port != 0) {
                reconnect = true;
                reconnectReason = reason;
                reconnectIsError = isError;
                host = null;
                port = 0;
                selector.wakeup();
            }
        }
    }

    private void process() {
        while (!thread.isInterrupted()) {
            try {
                doReconnect();
                doConnect();
                updateWriteInterestOps();
                doTransceive();
            } catch (final EOFException ex) {
                final String tmp = ex.getMessage();
                final String message = tmp == null ? "EOF" : tmp;
                if (debugProtocol != null) {
                    debugProtocol.debugProtocolWrite("socket:exception "+message, ex);
                }
                processDisconnect(message, false);
            } catch (final IOException ex) {
                final String tmp = ex.getMessage();
                final String message = tmp == null ? "I/O error" : tmp;
                if (debugProtocol != null) {
                    debugProtocol.debugProtocolWrite("socket:exception "+message, ex);
                }
                processDisconnect(message, true);
            }
        }
    }

    private void doConnect() throws IOException {
        final boolean notifyConnected;
        synchronized (syncOutput) {
            if (isConnected || socketChannel == null) {
                notifyConnected = false;
            } else {
                isConnected = socketChannel.finishConnect();
                if (isConnected) {
                    interestOps = SelectionKey.OP_READ;
                    updateInterestOps();
                    notifyConnected = true;
                } else {
                    notifyConnected = false;
                }
            }
        }
        if (notifyConnected) {
            for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
                clientSocketListener.connected();
            }
        }
    }

    private void doReconnect() throws IOException {
        assert Thread.currentThread() == thread;

        final boolean doReconnect;
        final boolean doDisconnect;
        @Nullable final String disconnectReason;
        final boolean disconnectIsError;
        synchronized (syncConnect) {
            if (reconnect) {
                reconnect = false;
                if (host != null && port != 0) {
                    doReconnect = true;
                    doDisconnect = false;
                    disconnectReason = "reconnect to "+host+":"+port;
                    disconnectIsError = false;
                } else {
                    doReconnect = false;
                    doDisconnect = true;
                    disconnectReason = reconnectReason;
                    disconnectIsError = reconnectIsError;
                }
            } else {
                doReconnect = false;
                doDisconnect = false;
                disconnectReason = null;
                disconnectIsError = false;
            }
        }
        if (doReconnect) {
            assert disconnectReason != null;
            processDisconnect(disconnectReason, disconnectIsError);
            final String connectHost;
            final int connectPort;
            synchronized (syncConnect) {
                disconnectPending = true;
                connectHost = host;
                connectPort = port;
            }
            if (connectHost != null) {
                processConnect(connectHost, connectPort);
            }
        }
        if (doDisconnect) {
            processDisconnect(disconnectReason, disconnectIsError);
        }
    }

    private void doTransceive() throws IOException {
        selector.select();
        final Collection<SelectionKey> selectedKeys = selector.selectedKeys();
        if (selectedKeys.remove(selectionKey) && isConnected) {
            processRead();
            processWrite();
        }
        assert selectedKeys.isEmpty();
    }

    private void processConnect(@NotNull final String host, final int port) throws IOException {
        assert Thread.currentThread() == thread;

        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:connecting to "+host+":"+port);
        }
        for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
            clientSocketListener.connecting();
        }

        final SocketAddress socketAddress = new InetSocketAddress(host, port);
        synchronized (syncOutput) {
            outputBuffer.clear();
            inputBuffer.clear();
            selectionKey = null;
            try {
                socketChannel = SocketChannel.open();
                selectableChannel = socketChannel.configureBlocking(false);
                try {
                    isConnected = socketChannel.connect(socketAddress);
                } catch (final UnresolvedAddressException ex) {
                    //noinspection ObjectToString
                    throw new IOException("Cannot resolve address: "+socketAddress, ex);
                } catch (final IllegalArgumentException ex) {
                    throw new IOException(ex.getMessage(), ex);
                }
                try {
                    socketChannel.socket().setTcpNoDelay(true);
                } catch (final SocketException ex) {
                    if (debugProtocol != null) {
                        debugProtocol.debugProtocolWrite("socket:cannot set TCP_NODELAY option: "+ex.getMessage());
                    }
                }
                interestOps = SelectionKey.OP_CONNECT;
                selectionKey = selectableChannel.register(selector, interestOps);
            } finally {
                if (selectionKey == null) {
                    socketChannel = null;
                    selectableChannel = null;
                    isConnected = false;
                    interestOps = 0;
                }
            }
        }
    }

    private void processDisconnect(@NotNull final String reason, final boolean isError) {
        assert Thread.currentThread() == thread;

        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:disconnecting: "+reason+(isError ? " [unexpected]" : ""));
        }
        final boolean notifyListeners;
        synchronized (syncConnect) {
            notifyListeners = disconnectPending;
            disconnectPending = false;
        }
        if (notifyListeners) {
            model.getGuiStateManager().disconnecting(reason, isError);
            for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
                clientSocketListener.disconnecting(reason, isError);
            }
        }

        try {
            synchronized (syncOutput) {
                if (selectionKey != null) {
                    selectionKey.cancel();
                    selectionKey = null;
                    outputBuffer.clear();

                    try {
                        if (socketChannel != null) {
                            socketChannel.socket().shutdownOutput();
                        }
                    } catch (final IOException ignored) {
                        // ignore
                    }
                    try {
                        if (socketChannel != null) {
                            socketChannel.close();
                        }
                    } catch (final IOException ignored) {
                        // ignore
                    }
                    socketChannel = null;
                    selectableChannel = null;
                    inputBuffer.clear();
                }
            }
        } finally {
            if (notifyListeners) {
                model.getGuiStateManager().disconnected();
                for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
                    clientSocketListener.disconnected(reason);
                }
            }
        }
    }

    private void processRead() throws IOException {
        synchronized (syncOutput) {
            if (socketChannel == null) {
                return;
            }

            if (socketChannel.read(inputBuffer) == -1) {
                throw new EOFException();
            }
        }
        inputBuffer.flip();
        processReadCommand();
        inputBuffer.compact();
    }

    private void processReadCommand() {
        while (true) {
            if (inputLen == -1) {
                if (inputBuffer.remaining() < 2) {
                    break;
                }

                inputLen = (inputBuffer.get()&0xFF)*0x100+(inputBuffer.get()&0xFF);
            }

            if (inputBuffer.remaining() < inputLen) {
                break;
            }

            final int start = inputBuffer.position();
            final int end = start+inputLen;
            inputBuffer.position(start+inputLen);
            inputLen = -1;
            final ByteBuffer packet = ByteBuffer.wrap(inputBuf, start, end-start);
            packet.order(ByteOrder.BIG_ENDIAN);
            try {
                for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
                    clientSocketListener.packetReceived(packet);
                }
            } catch (final UnknownCommandException ex) {
                disconnect(ex.getMessage(), true);
                break;
            }
        }
    }

    public void writePacket(@NotNull final byte[] buf, final int len) {
        synchronized (syncOutput) {
            if (socketChannel == null) {
                return;
            }

            packetHeader[0] = (byte)(len/0x100);
            packetHeader[1] = (byte)len;
            try {
                try {
                    outputBuffer.put(packetHeader);
                    outputBuffer.put(buf, 0, len);
                } catch (final BufferOverflowException ex) {
                    throw new IOException("buffer overflow", ex);
                }
            } catch (final IOException ignored) {
                //noinspection UnusedCatchParameter
                try {
                    socketChannel.close();
                } catch (final IOException ignored2) {
                    // ignore
                }
                return;
            }
        }

        selector.wakeup();
        for (final ClientSocketListener clientSocketListener : clientSocketListeners.getListeners()) {
            clientSocketListener.packetSent(buf, len);
        }
    }

    private void processWrite() throws IOException {
        synchronized (syncOutput) {
            if (outputBuffer.remaining() <= 0) {
                return;
            }

            outputBuffer.flip();
            try {
                if (socketChannel != null) {
                    socketChannel.write(outputBuffer);
                } else {
                    outputBuffer.position(outputBuffer.limit());
                }
            } finally {
                outputBuffer.compact();
            }
        }
    }

    private void updateWriteInterestOps() {
        synchronized (syncOutput) {
            final int newInterestOps;
            if (outputBuffer.position() > 0) {
                newInterestOps = interestOps|SelectionKey.OP_WRITE;
            } else {
                newInterestOps = interestOps&~SelectionKey.OP_WRITE;
            }
            if (interestOps != newInterestOps) {
                interestOps = newInterestOps;
                updateInterestOps();
            }
        }
    }

    private void updateInterestOps() {
        if (debugProtocol != null) {
            debugProtocol.debugProtocolWrite("socket:set interest ops to "+interestOps);
        }
        assert Thread.holdsLock(syncOutput);
        if (selectionKey != null) {
            selectionKey.interestOps(interestOps);
        }
    }

}

---