source: java/main/src/main/java/com/framsticks/communication/ClientSideManagedConnection.java @ 99

package com.framsticks.communication;

import com.framsticks.communication.queries.ApplicationRequest;
import com.framsticks.communication.queries.CallRequest;
import com.framsticks.communication.queries.ProtocolRequest;
import com.framsticks.communication.queries.RegisterRequest;
import com.framsticks.communication.queries.UseRequest;
import com.framsticks.communication.queries.VersionRequest;
import com.framsticks.core.Path;
import com.framsticks.params.ListSource;
import com.framsticks.util.*;
import com.framsticks.util.dispatching.AtOnceDispatcher;
import com.framsticks.util.dispatching.Dispatcher;
import com.framsticks.util.dispatching.Dispatching;
import com.framsticks.util.dispatching.ExceptionResultHandler;
import com.framsticks.util.dispatching.Future;
import com.framsticks.util.dispatching.FutureHandler;
import com.framsticks.util.dispatching.JoinableState;
import com.framsticks.util.lang.Pair;
import com.framsticks.util.lang.Strings;
import com.framsticks.params.EventListener;

import org.apache.log4j.Logger;

import java.util.*;
import java.util.regex.Matcher;
import com.framsticks.util.dispatching.RunAt;

/**
 * @author Piotr Sniegowski
 */
public class ClientSideManagedConnection extends ManagedConnection {

        private final static Logger log = Logger.getLogger(ClientSideManagedConnection.class);

        private final List<Runnable> applicationRequestsBuffer = new LinkedList<>();
        private boolean isHandshakeDone = false;


        /**
         * @return the requestedVersion
         */
        public int getRequestedVersion() {
                return requestedVersion;
        }

        /**
         * @param requestedVersion the requestedVersion to set
         */
        public void setRequestedVersion(int requestedVersion) {
                this.requestedVersion = requestedVersion;
        }

        protected int requestedVersion = 4;

        public ClientSideManagedConnection() {
                setDescription("client connection");
                protocolVersion = -1;
        }


        private static abstract class InboundMessage {
                protected String currentFilePath;
                protected List<String> currentFileContent;
                protected final List<File> files = new ArrayList<File>();

                public abstract void eof();

                protected void initCurrentFile(String path) {
                        currentFileContent = new LinkedList<String>();
                        currentFilePath = path;
                }

                protected void finishCurrentFile() {
                        if (currentFileContent == null) {
                                return;
                        }
                        files.add(new File(currentFilePath, new ListSource(currentFileContent)));
                        currentFilePath = null;
                        currentFileContent = null;
                }

                public abstract void startFile(String path);

                public final void addLine(String line) {
                        assert line != null;
                        assert currentFileContent != null;
                        currentFileContent.add(line);
                }

                public List<File> getFiles() {
                        return files;
                }
        }

        protected List<String> readFileContent() {
                List<String> content = new LinkedList<String>();
                String line;
                while (!(line = getLine()).startsWith("eof")) {
                        content.add(line);
                }
                return content;
        }

        private static class SentQuery<C> extends InboundMessage {
                Request request;
                ClientSideResponseFuture callback;
                Dispatcher<C> dispatcher;

                public void startFile(String path) {
                        finishCurrentFile();
                        if (!Strings.notEmpty(path)) {
                                assert request instanceof ApplicationRequest;
                                path = ((ApplicationRequest) request).getPath();
                        }
                        Strings.assureNotEmpty(path);
                        initCurrentFile(path);
                }

                public void eof() {
                        assert Strings.notEmpty(currentFilePath);
                        finishCurrentFile();
                        //no-operation
                }

                @Override
                public String toString() {
                        return request.toString();
                }

                public void dispatchResponseProcess(final Response response) {
                        Dispatching.dispatchIfNotActive(dispatcher, new RunAt<C>(callback) {
                                @Override
                                protected void runAt() {
                                        callback.pass(response);
                                }
                        });
                }
        }

        private Map<Integer, SentQuery<?>> queryMap = new HashMap<>();

        private SentQuery<?> currentlySentQuery;

        public void send(ProtocolRequest request, ClientSideResponseFuture callback) {
                //TODO RunAt
                sendImplementation(request, AtOnceDispatcher.getInstance(), callback);
        }



        public <C> void send(final ApplicationRequest request, final Dispatcher<C> dispatcher, final ClientSideResponseFuture callback) {
                synchronized (applicationRequestsBuffer) {
                        if (!isHandshakeDone) {
                                applicationRequestsBuffer.add(new Runnable() {
                                        @Override
                                        public void run() {
                                                sendImplementation(request, dispatcher, callback);
                                        }
                                });
                                return;
                        }
                }
                sendImplementation(request, dispatcher, callback);
        }

        private <C> void sendImplementation(Request request, Dispatcher<C> dispatcher, ClientSideResponseFuture callback) {
                callback.setRequest(request);

                if (getState().ordinal() > JoinableState.RUNNING.ordinal()) {
                        log.fatal("not connected");
                        return;
                }

                final SentQuery<C> sentQuery = new SentQuery<C>();
                sentQuery.request = request;
                sentQuery.callback = callback;
                sentQuery.dispatcher = dispatcher;

                senderThread.dispatch(new RunAt<Connection>(callback) {
                        @Override
                        protected void runAt() {
                                Integer id;
                                synchronized (ClientSideManagedConnection.this) {

                                        while (!(requestIdEnabled || currentlySentQuery == null)) {
                                                try {
                                                        ClientSideManagedConnection.this.wait();
                                                } catch (InterruptedException ignored) {
                                                        break;
                                                }
                                        }
                                        if (requestIdEnabled) {
                                                queryMap.put(nextQueryId, sentQuery);
                                                id = nextQueryId++;
                                        } else {
                                                currentlySentQuery = sentQuery;
                                                id = null;
                                        }
                                }
                                String command = sentQuery.request.getCommand();
                                StringBuilder message = new StringBuilder();
                                message.append(command);
                                if (id != null) {
                                        message.append(" ").append(id);
                                }
                                message.append(" ");
                                sentQuery.request.construct(message);
                                String out = message.toString();

                                putLine(out);
                                flushOut();
                                log.debug("sending query: " + out);

                        }
                });
                /*
                synchronized (this) {
                        log.debug("queueing query: " + query);
                        queryQueue.offer(sentQuery);
                        notifyAll();
                }
                 */
        }

        @Override
        public String toString() {
                return "client connection " + address;
        }


        private void sendQueryVersion(final int version, final Future<Void> future) {
                send(new VersionRequest().version(version), new ClientSideResponseFuture(future) {
                        @Override
                        protected void processOk(Response response) {
                                protocolVersion = version;
                                if (version < requestedVersion) {
                                        /** it is an implicit loop here*/
                                        sendQueryVersion(version + 1, future);
                                        return;
                                }
                                send(new UseRequest().feature("request_id"), new ClientSideResponseFuture(future) {

                                        @Override
                                        protected void processOk(Response response) {
                                                requestIdEnabled = true;
                                                future.pass(null);
                                        }
                                });

                        }
                });
        }

        private synchronized SentQuery<?> fetchQuery(Integer id, boolean remove) {
                if (id == null) {
                        if (requestIdEnabled) {
                                return null;
                        }
                        SentQuery<?> result = currentlySentQuery;
                        if (remove) {
                                currentlySentQuery = null;
                                notifyAll();
                        }
                        return result;
                }
                if (queryMap.containsKey(id)) {
                        SentQuery<?> result = queryMap.get(id);
                        if (remove) {
                                queryMap.remove(id);
                        }
                        return result;
                }
                return null;
        }

        private int nextQueryId = 0;

        protected void processMessage(InboundMessage inboundMessage) {
                if (inboundMessage == null) {
                        log.error("failed to use any inbound message");
                        return;
                }

                String line;
                while (!(line = getLine()).startsWith("eof")) {
                        // log.debug("line: " + line);
                        inboundMessage.addLine(line);
                }
                inboundMessage.eof();
        }

        protected void processEvent(String rest) {
                Matcher matcher = Request.EVENT_PATTERN.matcher(rest);
                if (!matcher.matches()) {
                        throw new FramsticksException().msg("invalid event line").arg("rest", rest);
                }
                String fileLine = getLine();
                if (!fileLine.equals("file")) {
                        throw new FramsticksException().msg("expected file line").arg("got", fileLine);
                }
                String eventObjectPath = Request.takeGroup(rest, matcher, 1).toString();
                String eventCalleePath = Request.takeGroup(rest, matcher, 2).toString();
                final File file = new File("", new ListSource(readFileContent()));
                log.debug("firing event " + eventObjectPath);
                EventListener<File> listener;
                synchronized (registeredListeners) {
                        listener = registeredListeners.get(eventObjectPath);
                }
                if (listener  == null) {
                        throw new FramsticksException().msg("failed to find registered event").arg("event path", eventObjectPath).arg("object", eventCalleePath);
                }
                listener.action(file);
        }

        protected void processMessageStartingWith(String line) {
                try {
                        Pair<CharSequence, CharSequence> command = Request.takeIdentifier(line);
                        if (command.first.equals("event")) {
                                processEvent(command.second.toString());
                                return;
                        }
                        Pair<Integer, CharSequence> rest = takeRequestId(command.second);

                        if (command.first.equals("file")) {
                                SentQuery<?> sentQuery = fetchQuery(rest.first, false);
                                sentQuery.startFile(rest.second.toString());
                                processMessage(sentQuery);
                                return;
                        }

                        SentQuery<?> sentQuery = fetchQuery(rest.first, true);
                        if (sentQuery == null) {
                                return;
                        }
                        log.debug("parsing response for request " + sentQuery);

                        sentQuery.dispatchResponseProcess(new Response(command.first.equals("ok"), rest.second.toString(), sentQuery.getFiles()));
                } catch (FramsticksException e) {
                        throw new FramsticksException().msg("failed to process message").arg("starting with line", line).cause(e);
                }
        }

        protected final ExceptionResultHandler closeOnFailure = new ExceptionResultHandler() {

                @Override
                public void handle(FramsticksException exception) {
                        interrupt();
                        // finish();
                }
        };

        @Override
        protected void receiverThreadRoutine() {
                startClientConnection(this);

                sendQueryVersion(1, new FutureHandler<Void>(closeOnFailure) {

                        @Override
                        protected void result(Void result) {
                                synchronized (applicationRequestsBuffer) {
                                        isHandshakeDone = true;
                                        for (Runnable r : applicationRequestsBuffer) {
                                                r.run();
                                        }
                                        applicationRequestsBuffer.clear();
                                }
                        }
                });

                processInputBatchesUntilClosed();
        }

        protected void processNextInputBatch() {
                processMessageStartingWith(getLine());
        }

        protected final Map<String, EventListener<File>> registeredListeners = new HashMap<>();

        public <C> void addListener(String path, final EventListener<File> listener, final Dispatcher<C> dispatcher, final Future<Void> future) {
                send(new RegisterRequest().path(path), dispatcher, new ClientSideResponseFuture(future) {
                        @Override
                        protected void processOk(Response response) {
                                synchronized (registeredListeners) {
                                        registeredListeners.put(Path.validateString(response.getComment()), listener);
                                }
                                future.pass(null);
                        }
                });
        }

        public <C> void removeListener(EventListener<File> listener, final Dispatcher<C> dispatcher, final Future<Void> future) {
                String eventPath = null;
                synchronized (registeredListeners) {
                        for (Map.Entry<String, EventListener<File>> e : registeredListeners.entrySet()) {
                                if (e.getValue() == listener) {
                                        eventPath = e.getKey();
                                        break;
                                }
                        }
                }
                if (eventPath == null) {
                        future.handle(new FramsticksException().msg("listener is not registered").arg("listener", listener));
                        return;
                }

                final String finalEventPath = eventPath;
                                //TODO add arguments to the exception
                send(new CallRequest().procedure("remove").path(eventPath), dispatcher, new ClientSideResponseFuture(future) {

                        @Override
                        protected void processOk(Response response) {
                                synchronized (registeredListeners) {
                                        registeredListeners.remove(finalEventPath);
                                }
                                future.pass(null);
                        }
                });
        }
}