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

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.params.*;
import com.framsticks.params.EventListener;
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.structure.AbstractTree;
import com.framsticks.structure.Path;
import com.framsticks.structure.messages.ListChange;
import com.framsticks.structure.messages.ValueChange;
import com.framsticks.util.*;
import com.framsticks.util.dispatching.AtOnceDispatcher;
import com.framsticks.util.dispatching.Dispatching;
import com.framsticks.util.dispatching.DispatchingFuture;
import com.framsticks.util.dispatching.FutureHandler;
import com.framsticks.util.dispatching.Future;
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.dispatching.RunAt;

import static com.framsticks.structure.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() {
                bufferedDispatcher.setBuffer(true);
                registry.registerAndBuild(ListChange.class);
                registry.registerAndBuild(ValueChange.class);
        }

        public void setAddress(Address address) {
                this.connection = Connection.to(new ClientSideManagedConnection(), address);
                this.connection.setExceptionHandler(this);
                this.connection.setNeedFileAcceptor(this);
        }

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

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

        public final ClientSideManagedConnection getConnection() {
                return connection;
        }


        @Override
        public String toString() {
                return super.toString() + "[" + getAddress() + "]";
        }

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

                        @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();
        //      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<FutureHandler<FramsClass>>> infoRequests = new HashMap<String, Set<FutureHandler<FramsClass>>>();


        /** Fetch information.
         *
         * This method does not check whether the info is already in the cache, it will always issue a request or join
         * with other already issued but returned info request. Too issue request only when needed, use TreeOperations.findInfo()
         *
         * */
        @Override
        public void info(final Path path, final FutureHandler<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<FutureHandler<FramsClass>> futures = new HashSet<FutureHandler<FramsClass>>();
                futures.add(future);
                infoRequests.put(name, futures);

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

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

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

                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 FutureHandler<Path> future) {
                assert isActive();
                final ExceptionHandler remover = pathRemoveHandler(path, future);

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

                                final Access access = registry.prepareAccess(path.getTop().getParam(), false);
                                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 FutureHandler<Integer> future) {
                assert isActive();
                final Integer flag = bindAccess(path).set(param, value);

                log.trace("storing value {} for {}", param, path);

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

                bufferedDispatcher.getTargetDispatcher().dispatch(new RunAt<RemoteTree>(ThrowExceptionHandler.getInstance()) {
                        @Override
                        protected void runAt() {

                                connection.send(new InfoRequest().path("/"), bufferedDispatcher.getTargetDispatcher(), new ClientSideResponseFuture(this) {
                                        @Override
                                        protected void processOk(Response response) {
                                                assert bufferedDispatcher.getTargetDispatcher().isActive();
                                                FramsClass framsClass = Loaders.loadFramsClass(response.getFiles().get(0).getContent());
                                                putInfoIntoCache(framsClass);

                                                assignRootParam(Param.build().name("Tree").id(RemoteTree.this.getName()).type("o " + framsClass.getId()).finish(CompositeParam.class));
                                                bufferedDispatcher.setBuffer(false);
                                        }
                                });

                        }
                });
        }

        @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 <R> void call(@Nonnull final Path path, @Nonnull final ProcedureParam procedure, @Nonnull Object[] arguments, final Class<R> resultType, final FutureHandler<R> 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();

                                if (response.getFiles().size() == 0) {
                                        future.pass(null);
                                        return;
                                }
                                if (response.getFiles().size() == 1) {
                                        future.pass(AccessOperations.convert(resultType, response.getFiles().get(0), registry));
                                        return;
                                }
                                throw new FramsticksUnsupportedOperationException().msg("call result with multiple files");

                        }
                });

        }

        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 FutureHandler<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();

                                                listener.action(AccessOperations.convert(argumentType, file, registry));
                                        }
                                });
                        }
                };

                proxyListeners.put(listener, proxyListener);

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

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

}