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

package com.framsticks.communication;

import com.framsticks.params.Source;
import com.framsticks.params.annotations.AutoAppendAnnotation;
import com.framsticks.params.annotations.FramsClassAnnotation;
import com.framsticks.params.annotations.ParamAnnotation;
import com.framsticks.util.ExceptionHandler;
import com.framsticks.util.FramsticksException;
import com.framsticks.util.io.Encoding;
import com.framsticks.util.lang.Strings;

import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;
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.util.Collection;
import java.util.HashSet;
import java.util.Set;

import com.framsticks.util.dispatching.AbstractJoinable;
import com.framsticks.util.dispatching.Dispatcher;
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;
import com.framsticks.util.dispatching.ThrowExceptionHandler;

@FramsClassAnnotation
public abstract class Connection extends AbstractJoinable implements JoinableParent, ExceptionHandler {

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

        private PrintWriter output = null;
        private BufferedReader input = null;

        protected Socket socket = null;

        protected Address address;
        protected String description = "connection";

        protected final Thread<Connection> senderThread = new Thread<>();
        protected final Thread<Connection> receiverThread = new Thread<>();
        protected final JoinableCollection<Thread<Connection>> threads = new JoinableCollection<>();
        protected final Set<ConnectionListener> listeners = new HashSet<>();

        protected ExceptionHandler exceptionHandler = ThrowExceptionHandler.getInstance();

        /**
         *
         */
        public Connection() {
                threads.add(senderThread);
                threads.add(receiverThread);
        }

        protected void updateNames() {
                if (address == null) {
                        return;
                }
                senderThread.setName(description + " thread " + address + " sender");
                receiverThread.setName(description + " thread " + address + " receiver");
                threads.setObservableName(address + " connection threads");
        }

        public void setDescription(String description) {
                this.description = description;
                updateNames();
        }

        @AutoAppendAnnotation
        public Connection setAddress(Address address) {
                this.address = address;
                updateNames();
                return this;
        }

        @ParamAnnotation
        public Connection setAddress(String address) {
                return setAddress(new Address(address));
        }

        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;
                                        String line = lineBuffer.toString();

                                        synchronized (listeners) {
                                                for (ConnectionListener l : listeners) {
                                                        l.connectionIncomming(line);
                                                }
                                        }

                                        return line;
                                }
                                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;
                                readChars = input.read(readBuffer);
                                if (readChars < 0) {
                                        throw new SocketClosedException().msg("socket is closed");
                                }
                                iterator = 0;
                                bufferStart = 0;
                        }
                        throw new InterruptedException();
                } catch (Exception e) {
                        log.debug("failed to read line (closing): {}", e.getMessage());
                        throw new SocketClosedException().msg("failed to read line").cause(e);
                }
        }

        protected void putLine(String line) {
                synchronized (listeners) {
                        for (ConnectionListener l : listeners) {
                                l.connectionOutgoing(line);
                        }
                }
                output.println(line);
        }

        protected void flushOut() {
                output.flush();
        }

        protected abstract void processNextInputBatch();


        protected final void processInputBatchesUntilClosed() {
                while (isRunning() && !socket.isClosed()) {
                        try {
                                processNextInputBatch();
                        } catch (SocketClosedException e) {
                                log.log(isRunning() ? Level.ERROR : Level.DEBUG, "socket is closing: {}", e.getShortMessage(new StringBuilder()));
                                // log.log(isRunning() ? Level.ERROR : Level.DEBUG, "caught exception: ", e);
                                break;
                        } catch (FramsticksException e) {
                                log.debug("{} caught exception in receiver thread {}", this, e.getMessage());
                                handle(e);
                        } catch (Exception e) {
                                log.log(isRunning() ? Level.ERROR : Level.DEBUG, "caught exception: ", e);
                                break;
                        }
                }
                log.debug("{} finished processing input", this);
        }

        protected abstract void receiverThreadRoutine();

        // @SuppressWarnings("NN_NAKED_NOTIFY")
        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);
                }
        }

        @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 address != null ? description + " " + address : description;
        }

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

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

                receiverThread.dispatch(new RunAt<Connection>(this) {
                        @Override
                        protected void runAt() {
                                receiverThreadRoutine();
                                interruptJoinable();
                                // finishJoinable();
                        }
                });
        }

        @Override
        protected void joinableInterrupt() {
                Dispatching.drop(threads, this);
                finishJoinable();
        }

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

        protected static void startClientConnection(Connection connection) {
                while (connection.isRunning() && connection.socket == null) {
                        log.debug("connecting to {}", connection.address);
                        try {
                                connection.socket = new Socket(connection.getAddressObject().getHostName(), connection.getAddressObject().getPort());
                        } catch (IOException e) {
                                log.warn("{} failed to connect (retrying): {}", connection, e.getMessage());
                                Dispatching.sleep(0.5);
                        }
                }

                log.debug("{} connected", connection);
                try {
                        // connection.socket.setSoTimeout(500);
                        connection.setupStreams();
                } catch (Exception e) {
                        throw new FramsticksException().msg("failed to initialize socket").cause(e).arg("connection", connection);
                }
        }

        /**
         * @return the address
         */
        @ParamAnnotation
        public String getAddress() {
                return Strings.toStringNullProof(address, "?");
        }

        public Address getAddressObject() {
                return address;
        }

        /**
         * @return the listeners
         */
        public Collection<ConnectionListener> getListeners() {
                return listeners;
        }

        /**
         * @return the handler
         */
        public ExceptionHandler getExceptionHandler() {
                return exceptionHandler;
        }

        /**
         * @param handler the handler to set
         */
        public void setExceptionHandler(ExceptionHandler handler) {
                this.exceptionHandler = handler;
        }

        public static <T extends Connection> T to(T connection, Address address) {
                connection.setAddress(address);
                return connection;
        }

        @Override
        public void handle(FramsticksException exception) {
                log.debug("{} handling {}", this, exception.getMessage());
                exceptionHandler.handle(exception);
        }

        public Dispatcher<Connection> getReceiverDispatcher() {
                return receiverThread;
        }

        public Dispatcher<Connection> getSenderDispatcher() {
                return senderThread;
        }


        protected static String idToString(Integer id) {
                return id != null ? " " + id.toString() : "";
        }

        protected final void putFile(File file, Integer outId) {
                putLine("file" + idToString(outId)/* + " " + f.getPath()*/);
                Source content = file.getContent();
                String line;
                while ((line = content.readLine()) != null) {
                        putLine(line);
                }
                putLine("eof");
        }

        public final void sendFile(final String header, final File file, final Integer id, ExceptionHandler handler) {
                senderThread.dispatch(new RunAt<Connection>(handler) {
                        @Override
                        protected void runAt() {
                                if (header != null) {
                                        putLine(header);
                                }
                                putFile(file, id);
                                flushOut();
                        }
                });
        }

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

}