/*
 * Copyright 2000-2015 JetBrains s.r.o.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package jetbrains.buildServer.clouds.vmware.vmrun.remote.client;

import com.intellij.openapi.diagnostic.Logger;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collection;
import java.util.Collections;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import jetbrains.buildServer.clouds.vmware.vmrun.remote.RemoteTask;
import jetbrains.buildServer.clouds.vmware.vmrun.remote.RemoteTaskResult;
import jetbrains.buildServer.clouds.vmware.vmrun.remote.VMRunServer;
import jetbrains.buildServer.util.ExceptionUtil;
import jetbrains.buildServer.util.StringUtil;
import org.jetbrains.annotations.Nullable;

/**
 * @author Eugene Petrenko
 *         Created: 08.12.2009 12:55:04
 */
public class Client<T> {
  private static final Logger LOG = Logger.getInstance(Client.class.getName());

  private final T myDest;
  private final VMRunServer myServer;
  private final String myClientInfo = "client-" + StringUtil.generateUniqueHash();

  public Client(final T dest, final VMRunServer server, final ScheduledExecutorService service, int pingDelay) {
    myDest = dest;
    myServer = server;

    final Set<String> myPendingTasks = new CopyOnWriteArraySet<String>();

    service.scheduleWithFixedDelay(ExceptionUtil.catchAll("VMWare clien fetch commands", new Runnable() {
      public void run() {
        LOG.debug("Fetching tasks...");
        try {
          final Collection<RemoteTask> tasks = myServer.getTasks(myClientInfo);
          if (!tasks.isEmpty()) {
            LOG.info("Fetched " + tasks.size() + " tasks");
          }
          //TODO: cleanup queue if server has gone!
          for (final RemoteTask task : tasks) {
            service.execute(processTask(task));
          }
        } catch (Throwable t) {
          final String msg = t.getMessage();
          if (msg == null || !msg.contains("Connection refused")) {
            LOG.warn("Failed to fetch task list. " + t.getMessage(), t);
          } else {
            LOG.debug("Failed to fetch task list. " + t.getMessage(), t);
          }
        }
      }

      private Runnable processTask(final RemoteTask task) {
        return new Runnable() {
          @Nullable
          private Method findMethod() {
            for (Method method : myDest.getClass().getMethods()) {
              if (method.getName().equals(task.getMethodId())) {
                return method;
              }
            }
            return null;
          }

          private void postError(Throwable e) {
            post(new RemoteTaskResult(e, null, task.getRequestId()));
          }

          private void postResult(Object o) {
            post(new RemoteTaskResult(null, o, task.getRequestId()));
          }

          private void post(final RemoteTaskResult o) {
            new Runnable() {
              public void run() {
                try {
                  myServer.postResults(myClientInfo, Collections.singleton(o));
                  //Cleanup pending tasks with delay to overcome race conditions
                  service.schedule(new Runnable() {
                    public void run() {
                      myPendingTasks.remove(o.getRquestId());
                    }
                  }, 5, TimeUnit.SECONDS);
                } catch (Throwable t) {
                  LOG.warn("Failed to post result to queue: " + o + ", " + t.getMessage(), t);
                  service.schedule(this,1, TimeUnit.SECONDS);
                }
              }
            }.run();
          }

          public void run() {
            final String req = task.getRequestId();

            if (!myPendingTasks.add(req)) return;
            myPendingTasks.add(req);

            run2();
          }

          private void run2() {
            LOG.info("Processing task: " + task);

            final Method method = findMethod();
            if (method == null) {
              //noinspection ThrowableInstanceNeverThrown
              postError(new NoSuchMethodException("Object does not contain method: " + task.getMethodId()));
              return;
            }

            method.setAccessible(true);

            try {
              final Object result = method.invoke(myDest, task.getParameters());
              postResult(result);
            } catch (Throwable e) {
              final String msg = e.getMessage();
              LOG.warn("Failed to perform command. " + (msg == null ? "" : msg), e);
              postError(e instanceof InvocationTargetException ? ((InvocationTargetException)e).getTargetException() : e);
            }
          }
        };
      }
    }), pingDelay, pingDelay, TimeUnit.MILLISECONDS);
  }
}