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

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 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;

        public boolean requestIdEnabled = false;

        protected int protocolVersion = -1;

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

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

        protected void setUpThreadNames(String name) {
        }

        /**
         *
         */
        public Connection(String address, String description) {
                this.address = address;
                this.description = description;
                threads.setObservableName(address + " connection threads");
                threads.add(senderThread);
                threads.add(receiverThread);

                senderThread.setName(description + " thread " + address + " sender");
                receiverThread.setName(description + " thread " + address + " receiver");
        }

        public synchronized boolean isConnected() {
                return socket != null && socket.isConnected();
        }


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

        // // protected final Pair<String, String> breakLine(String line)
        // protected final Pair<Integer, String> parseRest(String rest) {
        //      Matcher matcher = (requestIdEnabled ? REQUEST_ID_ENABLED_PATTERN : REQUEST_ID_DISABLED_PATTERN).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));
        // }

        protected final Pair<Integer, CharSequence> takeRequestId(CharSequence line) {
                return Request.takeRequestId(requestIdEnabled, line);
        }

        static final int BUFFER_LENGTH = 1024;

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

        protected String getLine() {
                final StringBuilder lineBuffer = new StringBuilder();
                try {
                        while (!Thread.interrupted()) {
                                while (iterator < readChars) {
                                        if (readBuffer[iterator] != '\n') {
                                                ++iterator;
                                                continue;
                                        }
                                        /** Do not append new line. */
                                        lineBuffer.append(readBuffer, bufferStart, iterator - bufferStart);
                                        ++iterator;
                                        bufferStart = iterator;
                                        return lineBuffer.toString();
                                }
                                final int length = readChars - bufferStart;
                                if (length > 0) {
                                        assert bufferStart >= 0 && bufferStart < BUFFER_LENGTH;
                                        assert bufferStart + length <= BUFFER_LENGTH;
                                        lineBuffer.append(readBuffer, bufferStart, length);
                                }

                                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 setupStreams() {
                try {
                        output = new PrintWriter(new OutputStreamWriter(socket.getOutputStream(), Encoding.getFramsticksCharset()), true);
                        input = new BufferedReader(new InputStreamReader(socket.getInputStream(), Encoding.getFramsticksCharset()));
                        synchronized (this) {
                                this.notifyAll();
                        }
                } catch (IOException e) {
                        throw new FramsticksException().msg("failed to setup streams").cause(e).arg("connection", this);
                }
        }

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


        @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
        public String getName() {
                return description + " " + address;
        }

        @Override
        protected void joinableStart() {
                Dispatching.use(threads, this);

                senderThread.dispatch(new RunAt<Connection>() {
                        @Override
                        public void run() {
                                synchronized (Connection.this) {
                                        while (state.equals(JoinableState.RUNNING) && output == null) {
                                                Dispatching.wait(Connection.this, 500);
                                        }
                                }
                        }
                });

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

        @Override
        protected void joinableInterrupt() {
                protocolVersion = -1;
                Dispatching.drop(threads, this);
                finish();
        }


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

}