source: java/main/src/main/java/com/framsticks/remote/RemoteTree.java @ 100

package com.framsticks.remote;

import com.framsticks.communication.*;
import com.framsticks.communication.queries.CallRequest;
import com.framsticks.communication.queries.GetRequest;
import com.framsticks.communication.queries.InfoRequest;
import com.framsticks.communication.queries.SetRequest;
import com.framsticks.core.AbstractTree;
import com.framsticks.core.Path;
import com.framsticks.params.*;
import com.framsticks.params.EventListener;
import com.framsticks.params.annotations.AutoAppendAnnotation;
import com.framsticks.params.annotations.FramsClassAnnotation;
import com.framsticks.params.annotations.ParamAnnotation;
import com.framsticks.params.types.EventParam;
import com.framsticks.params.types.ProcedureParam;
import com.framsticks.parsers.Loaders;
import com.framsticks.parsers.MultiParamLoader;
import com.framsticks.core.Tree;
import com.framsticks.util.*;
import com.framsticks.util.dispatching.AtOnceDispatcher;
import com.framsticks.util.dispatching.Dispatching;
import com.framsticks.util.dispatching.Dispatching.DispatcherWaiter;
import com.framsticks.util.dispatching.DispatchingFuture;
import com.framsticks.util.dispatching.ExceptionResultHandler;
import com.framsticks.util.dispatching.Future;
import com.framsticks.util.dispatching.FutureHandler;
import com.framsticks.util.dispatching.Joinable;
import com.framsticks.util.dispatching.JoinableParent;
import com.framsticks.util.dispatching.JoinableState;
import com.framsticks.util.dispatching.ThrowExceptionHandler;
import com.framsticks.util.lang.Casting;
import com.framsticks.util.dispatching.RunAt;
import static com.framsticks.core.TreeOperations.*;

import java.util.*;

import javax.annotation.Nonnull;

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

/**
 * @author Piotr Sniegowski
 */
@FramsClassAnnotation
public final class RemoteTree extends AbstractTree implements JoinableParent {

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

        protected ClientSideManagedConnection connection;

        public RemoteTree() {
        }

        @ParamAnnotation
        public void setAddress(String address) {
                setConnection(Connection.to(new ClientSideManagedConnection(), new Address(address)));
        }

        @ParamAnnotation
        public String getAddress() {
                return connection == null ? "<disconnected>" : connection.getAddress().toString();
        }

        @AutoAppendAnnotation
        public void setConnection(final ClientSideManagedConnection connection) {
                this.connection = connection;
                this.connection.setExceptionHandler(this);
        }

        public final ClientSideManagedConnection getConnection() {
                return connection;
        }

        @Override
        public String toString() {
                assert Dispatching.isThreadSafe();
                return "remote tree " + getName() + "(" + getAddress() + ")";
        }

        protected ExceptionResultHandler pathRemoveHandler(final Path path, final ExceptionResultHandler handler) {
                return new ExceptionResultHandler() {

                        @Override
                        public void handle(final FramsticksException exception) {
                                Dispatching.dispatchIfNotActive(RemoteTree.this, new RunAt<RemoteTree>(RemoteTree.this) {

                                        @Override
                                        protected void runAt() {
                                                assert path.getTree().isActive();
                                                log.info("path is invalid (removing): {}", path);
                                                bindAccess(path.getUnder()).set(path.getTop().getParam(), null);
                                                handler.handle(exception);
                                        }
                                });
                        }
                };
        }

        // @Override
        // public void get(final Path path, final ValueParam param, final Future<Object> future) {
        //      assert isActive();
        //      assert param != null;
        //      // assert path.isResolved();
        //      //TODO only do that if needed
        //      connection.send(new GetRequest().field(param.getId()).path(path.getTextual()), this, new ClientSideResponseFuture(pathRemoveHandler(path, future)) {
        //              @Override
        //              protected void processOk(Response response) {
        //                      assert isActive();
        //                      processFetchedValues(path, response.getFiles());
        //                      future.pass(bindAccess(path.tryResolveIfNeeded()).get(param, Object.class));
        //              }
        //      });
        // }

        protected final Map<String, Set<Future<FramsClass>>> infoRequests = new HashMap<String, Set<Future<FramsClass>>>();


        @Override
        public void info(final Path path, final Future<FramsClass> future) {

                final String name = path.getTop().getParam().getContainedTypeName();

                if (infoRequests.containsKey(name)) {
                        infoRequests.get(name).add(future);
                        return;
                }

                log.debug("issuing info request for {}", name);
                final Set<Future<FramsClass>> futures = new HashSet<Future<FramsClass>>();
                futures.add(future);
                infoRequests.put(name, futures);

                final Future<FramsClass> compositeFuture = DispatchingFuture.create(this, new Future<FramsClass>() {

                        @Override
                        public void handle(FramsticksException exception) {
                                assert isActive();
                                infoRequests.remove(name);
                                for (Future<FramsClass> f : futures) {
                                        f.handle(exception);
                                }
                        }

                        @Override
                        protected void result(FramsClass framsClass) {
                                assert isActive();
                                putInfoIntoCache(framsClass);
                                infoRequests.remove(name);
                                for (Future<FramsClass> f : futures) {
                                        f.pass(framsClass);
                                }
                        }
                });

                //TODO: if the info is in the cache, then don't communicate
                connection.send(new InfoRequest().path(path.getTextual()), AtOnceDispatcher.getInstance(), new ClientSideResponseFuture(compositeFuture) {
                        @Override
                        protected void processOk(Response response) {
                                assert connection.getReceiverDispatcher().isActive();

                                if (response.getFiles().size() != 1) {
                                        throw new FramsticksException().msg("invalid number of files in response").arg("files", response.getFiles().size());
                                }
                                if (!path.isTheSame(response.getFiles().get(0).getPath())) {
                                        throw new FramsticksException().msg("path mismatch").arg("returned path", response.getFiles().get(0).getPath());
                                }
                                FramsClass framsClass = Loaders.loadFramsClass(response.getFiles().get(0).getContent());

                                CompositeParam thisParam = path.getTop().getParam();
                                if (!thisParam.isMatchingContainedName(framsClass.getId())) {
                                        throw new FramsticksException().msg("framsclass id mismatch").arg("request", thisParam.getContainedTypeName()).arg("fetched", framsClass.getId());
                                }
                                compositeFuture.pass(framsClass);
                        }
                });
        }

        @Override
        public void get(final Path path, final Future<Path> future) {
                assert isActive();
                final ExceptionResultHandler remover = pathRemoveHandler(path, future);

                log.trace("fetching values for {}", path);
                findInfo(path, new FutureHandler<FramsClass>(remover) {
                        @Override
                        protected void result(FramsClass result) {

                                final Access access = registry.prepareAccess(path.getTop().getParam());
                                connection.send(new GetRequest().path(path.getTextual()), AtOnceDispatcher.getInstance(), new ClientSideResponseFuture(remover) {
                                        @Override
                                        protected void processOk(Response response) {
                                                processFetchedValues(path, response.getFiles(), access, future);
                                        }
                                });
                        }
                });
        }

        @Override
        public void set(final Path path, final PrimitiveParam<?> param, final Object value, final Future<Integer> future) {
                assert isActive();
                final Integer flag = bindAccess(path).set(param, value);

                log.trace("storing value {} for {}", param, path);
                //TODO break in passing exception handler is here
                connection.send(new SetRequest().value(value.toString()).field(param.getId()).path(path.getTextual()), this, new ClientSideResponseFuture(future) {
                        @Override
                        protected void processOk(Response response) {
                                future.pass(flag);
                        }
                });
        }

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

                dispatch(new RunAt<RemoteTree>(ThrowExceptionHandler.getInstance()) {
                        @Override
                        protected void runAt() {
                                final DispatcherWaiter<Tree, Tree> waiter = new DispatcherWaiter<Tree, Tree>(RemoteTree.this);

                                connection.send(new InfoRequest().path("/"), waiter, new ClientSideResponseFuture(this) {
                                        @Override
                                        protected void processOk(Response response) {
                                                FramsClass framsClass = processFetchedInfo(RemoteTree.this, response.getFiles().get(0));
                                                assignRootParam(Param.build().name("Tree").id(RemoteTree.this.getName()).type("o " + framsClass.getId()).finish(CompositeParam.class));
                                        }
                                });

                                waiter.waitFor();
                        }
                });
        }

        @Override
        protected void joinableInterrupt() {
                Dispatching.drop(connection, this);
                super.joinableInterrupt();
        }

        @Override
        protected void joinableFinish() {
                super.joinableFinish();

        }

        @Override
        public void joinableJoin() throws InterruptedException {
                Dispatching.join(connection);
                super.joinableJoin();
        }

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

        @Override
        public void call(@Nonnull final Path path, @Nonnull final ProcedureParam procedure, @Nonnull Object[] arguments, final Future<Object> future) {
                assert isActive();
                assert path.isResolved();

                //TODO validate arguments type using params
                connection.send(new CallRequest().procedure(procedure.getId()).arguments(Arrays.asList(arguments)).path(path.getTextual()), this, new ClientSideResponseFuture(future) {
                        @Override
                        protected void processOk(Response response) {
                                assert isActive();
                                // InstanceUtils.processFetchedValues(path, response.getFiles());
                                future.pass(null);
                        }
                });

        }

        protected final Map<EventListener<?>, EventListener<File>> proxyListeners = new IdentityHashMap<>();

        public <A> void addListener(Path path, EventParam param, final EventListener<A> listener, final Class<A> argumentType, final Future<Void> future) {
                assert isActive();
                assert path.isResolved();
                if ((!argumentType.equals(Object.class)) && (null == registry.getFramsClassForJavaClass(argumentType))) {
                        registry.registerAndBuild(argumentType);
                }

                final EventListener<File> proxyListener = new EventListener<File>() {

                        @Override
                        public void action(final File file) {
                                Dispatching.dispatchIfNotActive(RemoteTree.this, new RunAt<RemoteTree>(RemoteTree.this) {

                                        @Override
                                        protected void runAt() {
                                                assert isActive();
                                                if (argumentType.equals(Object.class)) {
                                                        listener.action(Casting.tryCast(argumentType, file));
                                                        return;
                                                }
                                                Access access = registry.createAccess(argumentType);
                                                Object argument = access.createAccessee();
                                                access.select(argument);
                                                if (!argumentType.isInstance(argument)) {
                                                        throw new FramsticksException().msg("created argument is of wrond type").arg("expected", argumentType).arg("created", argument.getClass());
                                                }
                                                A typedArgument = argumentType.cast(argument);

                                                // log.info("executing event with argument {}", argumentType);
                                                MultiParamLoader loader = new MultiParamLoader();
                                                loader.setNewSource(file.getContent());
                                                loader.addBreakCondition(MultiParamLoader.Status.AfterObject);
                                                loader.addAccess(access);
                                                loader.go();

                                                listener.action(typedArgument);
                                        }
                                });
                        }
                };

                proxyListeners.put(listener, proxyListener);

                connection.addListener(Path.appendString(path.getTextual(), param.getId()), proxyListener, this, future);
        }

        public void removeListener(Path path, EventParam param, EventListener<?> listener, Future<Void> future) {
                assert isActive();
                EventListener<File> proxyListener = proxyListeners.get(listener);
                connection.removeListener(proxyListener, this, future);
        }

}