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

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.Casting;
import com.framsticks.util.lang.Pair;
import com.framsticks.util.lang.Strings;
import com.framsticks.params.EventListener;

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

import java.util.*;
import java.util.regex.Matcher;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;

import com.framsticks.util.dispatching.RunAt;

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

        private final static Logger log = LogManager.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;
        }



        protected List<String> readFileContent() {
                List<String> content = new LinkedList<String>();
                String line;
                boolean longValue = false;
                while (true) {
                        line = getLine();
                        if (longValue) {
                                if (line.endsWith("~") && !line.endsWith("\\~")) {
                                        longValue = false;
                                }
                        } else {
                                if (line.equals("eof")) {
                                        break;
                                }
                                if (line.endsWith(":~")) {
                                        longValue = true;
                                }
                        }
                        content.add(line);
                }
                return content;
        }

        private static class SentQuery<C> {

                Request request;
                ClientSideResponseFuture callback;
                Dispatcher<C> dispatcher;
                protected final List<File> files = new ArrayList<File>();

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

                @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()) {
                        throw new FramsticksException().msg("connection is not connected").arg("connection", this);
                }

                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 @Nonnull SentQuery<?> fetchQuery(@Nullable Integer id, boolean remove) {
                try {
                        if (id == null) {
                                if (requestIdEnabled) {
                                        throw new FramsticksException().msg("request_id is enabled and id is missing");
                                }
                                SentQuery<?> result = currentlySentQuery;
                                if (remove) {
                                        currentlySentQuery = null;
                                        notifyAll();
                                }
                                return result;
                        }

                        if (!queryMap.containsKey(id)) {
                                throw new FramsticksException().msg("id is unknown").arg("id", id);
                        }

                        SentQuery<?> result = queryMap.get(id);
                        if (remove) {
                                queryMap.remove(id);
                        }
                        return result;

                } catch (FramsticksException e) {
                        throw new FramsticksException().msg("failed to match response to sent query").cause(e);
                }
        }

        private int nextQueryId = 0;

        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 processFile(Pair<Integer, CharSequence> rest) {
                final SentQuery<?> sentQuery = fetchQuery(rest.first, false);

                String currentFilePath = rest.second.toString();
                if (!Strings.notEmpty(currentFilePath)) {
                        currentFilePath = Casting.throwCast(ApplicationRequest.class, sentQuery.request).getPath();
                }

                sentQuery.files.add(new File(currentFilePath, new ListSource(readFileContent())));

        }

        protected void processMessageStartingWith(final String header) {
                try {
                        final Pair<CharSequence, CharSequence> command = Request.takeIdentifier(header);
                        if (command == null) {
                                throw new FramsticksException().msg("failed to parse command");
                        }
                        final CharSequence keyword = command.first;
                        if (keyword.equals("event")) {
                                processEvent(command.second.toString());
                                return;
                        }

                        final Pair<Integer, CharSequence> rest = takeRequestId(command.second);
                        if (rest == null) {
                                throw new FramsticksException().msg("failed to parse optional id and remainder");
                        }

                        if (keyword.equals("file")) {
                                processFile(rest);
                                return;
                        }
                        if (keyword.equals("ok") || keyword.equals("error")) {

                                final SentQuery<?> sentQuery = fetchQuery(rest.first, true);

                                log.debug("parsing response for request {}", sentQuery);

                                sentQuery.dispatchResponseProcess(new Response(command.first.equals("ok"), rest.second.toString(), sentQuery.getFiles()));
                                return;
                        }

                        throw new FramsticksException().msg("unknown command keyword").arg("keyword", keyword);
                } catch (FramsticksException e) {
                        throw new FramsticksException().msg("failed to process message").arg("starting with line", header).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);
                        }
                });
        }
}