source: java/main/src/main/java/com/framsticks/communication/Connection.java @ 90

package com.framsticks.communication;

import com.framsticks.util.FramsticksException;
import com.framsticks.util.io.Encoding;
import com.framsticks.util.lang.Pair;
import org.apache.log4j.Logger;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.framsticks.util.dispatching.AbstractJoinable;
import com.framsticks.util.dispatching.Dispatching;
import com.framsticks.util.dispatching.Joinable;
import com.framsticks.util.dispatching.JoinableCollection;
import com.framsticks.util.dispatching.JoinableParent;
import com.framsticks.util.dispatching.JoinableState;
import com.framsticks.util.dispatching.RunAt;
import com.framsticks.util.dispatching.Thread;

public abstract class Connection extends AbstractJoinable implements JoinableParent {

        protected final static Logger log = Logger.getLogger(Connection.class);

        protected PrintWriter output = null;
        protected BufferedReader input = null;

        protected Socket socket = null;

        protected volatile boolean connected = false;

        public boolean requestIdEnabled = false;

        protected int protocolVersion = -1;

        public String getAddress() {
                return address;
        }
        protected final String address;

        protected final Thread<Connection> senderThread = new Thread<>();
        protected final Thread<Connection> receiverThread = new Thread<>();
        protected final JoinableCollection<Thread<Connection>> threads = new JoinableCollection<>(true);

        /**
         *
         */
        public Connection(String address) {
                this.address = address;
                threads.add(senderThread);
                threads.add(receiverThread);
        }
        public boolean isConnected() {
                return connected;
        }


        protected static final String ARGUMENT_PATTERN_FRAGMENT = "((?:\\S+)|(?:\"[^\"]*\"))";
        protected static final Pattern requestIdEnabledPattern = Pattern.compile("^\\s*([0-9]+)(?:\\s+" + ARGUMENT_PATTERN_FRAGMENT + ")?\\n$");
        protected static final Pattern requestIDisabledPattern = Pattern.compile("^\\s*" + ARGUMENT_PATTERN_FRAGMENT + "?\\n$");
        protected static final Pattern eventPattern = Pattern.compile("^\\s*(\\S+)\\s*(\\S+)\\n");


        protected final Pair<Integer, String> parseRest(String rest) {
                Matcher matcher = (requestIdEnabled ? requestIdEnabledPattern : requestIDisabledPattern).matcher(rest);
                if (!matcher.matches()) {
                        log.fatal("unmatched first line of input: " + rest);
                        return null;
                }
                return new Pair<Integer, String>(requestIdEnabled ? Integer.parseInt(matcher.group(1)) : null, matcher.group(requestIdEnabled ? 2 : 1));
        }

        static final int BUFFER_LENGTH = 1024;

        int readChars = 0;
        int iterator = 0;
        int bufferStart = 0;
        char[] readBuffer = new char[BUFFER_LENGTH];

        protected String getLine() {
                StringBuilder lineBuffer = new StringBuilder();
                try {
                        while (!Thread.interrupted()) {
                                while (iterator < readChars) {
                                        if (readBuffer[iterator] != '\n') {
                                                ++iterator;
                                                continue;
                                        }
                                        lineBuffer.append(readBuffer, bufferStart, iterator - bufferStart + 1);
                                        ++iterator;
                                        bufferStart = iterator;
                                        return lineBuffer.toString();
                                }
                                lineBuffer.append(readBuffer, bufferStart, readChars - bufferStart);

                                readChars = 0;
                                while (readChars == 0) {
                                        try {
                                                readChars = input.read(readBuffer);
                                        } catch (SocketTimeoutException ignored) {
                                                //timeout - continue
                                        }
                                }
                                iterator = 0;
                                bufferStart = 0;
                        }
                        throw new InterruptedException();
                } catch (Exception e) {
                        throw new FramsticksException().msg("failed to read line").cause(e);
                }
        }

        protected abstract void receiverThreadRoutine();

        protected void setUpThread(Thread<Connection> thread, String name) {
                thread.setName("connection thread " + address + " " + name);
        }

        protected void runThreads() {
                try {
                        output = new PrintWriter(new OutputStreamWriter(socket.getOutputStream(), Encoding.getFramsticksCharset()), true);
                        input = new BufferedReader(new InputStreamReader(socket.getInputStream(), Encoding.getFramsticksCharset()));
                } catch (IOException e) {
                        log.error("buffer creation failure");
                        return;
                }

                setUpThread(senderThread, "sender");
                setUpThread(receiverThread, "receiver");
                Dispatching.use(threads, this);

                receiverThread.dispatch(new RunAt<Connection>() {
                        @Override
                        public void run() {
                                receiverThreadRoutine();
                        }
                });

        }

        /**
         * Returns Query associated with query getId.
         *
         * @return Query associated with query getId.
         */
        public int getProtocolVersion() {
                return protocolVersion;
        }

        @Override
        protected void joinableInterrupt() {
                protocolVersion = -1;

                connected = false;
                Dispatching.drop(threads, this);

                // finish();
        }

        @Override
        protected void joinableFinish() {
                try {
                        if (output != null) {
                                output.close();
                                output = null;
                        }

                        if (input != null) {
                                input.close();
                                input = null;
                        }


                        if (socket != null) {
                                socket.close();
                                socket = null;
                        }
                } catch (Exception e) {
                        log.error("failed to stop connection: ", e);
                }
                log.debug("connection closed");
        }

        @Override
        public void childChangedState(Joinable joinable, JoinableState state) {
                proceedToState(state);
        }

        @Override
        protected void joinableJoin() throws InterruptedException {
                Dispatching.join(threads);

        }

}