/*
 * Copyright (c) 2008-2017, Hazelcast, Inc. All Rights Reserved.
 *
 * 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 com.hazelcast.jet.impl;

import com.hazelcast.client.impl.HazelcastClientInstanceImpl;
import com.hazelcast.client.impl.protocol.ClientMessage;
import com.hazelcast.client.impl.protocol.codec.JetCancelJobCodec;
import com.hazelcast.client.impl.protocol.codec.JetCompleteResourceCodec;
import com.hazelcast.client.impl.protocol.codec.JetExecuteJobCodec;
import com.hazelcast.client.impl.protocol.codec.JetUpdateResourceCodec;
import com.hazelcast.client.spi.impl.ClientInvocation;
import com.hazelcast.client.spi.impl.ClientInvocationFuture;
import com.hazelcast.core.ExecutionCallback;
import com.hazelcast.jet.DAG;
import com.hazelcast.jet.config.JetConfig;
import com.hazelcast.jet.JetInstance;
import com.hazelcast.jet.Job;
import com.hazelcast.jet.config.JobConfig;
import com.hazelcast.jet.config.ResourceConfig;
import com.hazelcast.jet.impl.deployment.ResourceIterator;
import com.hazelcast.jet.impl.util.Util;
import com.hazelcast.logging.ILogger;
import com.hazelcast.nio.serialization.Data;
import com.hazelcast.util.function.Supplier;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static java.util.stream.Collectors.toList;

public class JetClientInstanceImpl extends AbstractJetInstance {

    private final HazelcastClientInstanceImpl client;
    private final ILogger logger;

    public JetClientInstanceImpl(HazelcastClientInstanceImpl hazelcastInstance) {
        super(hazelcastInstance);
        this.client = hazelcastInstance;
        this.logger = hazelcastInstance.getLoggingService().getLogger(JetInstance.class);
    }

    @Override
    public JetConfig getConfig() {
        throw new UnsupportedOperationException("Jet Configuration is not available on the client");
    }

    @Override
    public Job newJob(DAG dag) {
        return new JobImpl(dag);
    }

    @Override
    public Job newJob(DAG dag, JobConfig config) {
        return new JobImpl(dag, config);
    }

    private class JobImpl implements Job {

        private final DAG dag;
        private final JobConfig config;

        protected JobImpl(DAG dag) {
            this(dag, new JobConfig());
        }

        protected JobImpl(DAG dag, JobConfig config) {
            this.dag = dag;
            this.config = config;
        }

        @Override
        public Future<Void> execute() {
            long executionId = getIdGenerator().newId();
            deployResources(executionId);
            Data dagData = client.getSerializationService().toData(dag);
            int partitionId = client.getPartitionService().getPartition(executionId).getPartitionId();
            ClientInvocation invocation = new ClientInvocation(client,
                    JetExecuteJobCodec.encodeRequest(executionId, dagData), partitionId);
            return new ExecutionFuture(invocation.invoke(), executionId, partitionId);
        }

        private void deployResources(long executionId) {
            final Set<ResourceConfig> resources = config.getResourceConfigs();
            if (logger.isFineEnabled() && resources.size() > 0) {
                logger.fine("Deploying the following resources for " + executionId + ":" + resources);
            }
            new ResourceIterator(resources, config.getResourcePartSize()).forEachRemaining(part -> {
                Data partData = client.getSerializationService().toData(part);
                invokeOnCluster(() -> JetUpdateResourceCodec.encodeRequest(executionId, partData));
            });
            resources.forEach(r -> {
                Data descriptorData = client.getSerializationService().toData(r.getDescriptor());
                invokeOnCluster(() -> JetCompleteResourceCodec.encodeRequest(executionId, descriptorData));
            });
            logger.fine("Resource deployment for job " + executionId + " completed.");
        }

        private List<ClientMessage> invokeOnCluster(Supplier<ClientMessage> messageSupplier) {
            return client.getCluster().getMembers().stream()
                         .map(m -> new ClientInvocation(client, messageSupplier.get(), m.getAddress()).invoke())
                         .collect(toList())
                         .stream()
                         .map(Util::uncheckedGet)
                         .collect(toList());
        }


    }

    private final class ExecutionFuture implements Future<Void> {

        private final ClientInvocationFuture future;
        private final long executionId;
        private final int partitionId;

        protected ExecutionFuture(ClientInvocationFuture future, long executionId, int partitionId) {
            this.future = future;
            this.executionId = executionId;
            this.partitionId = partitionId;
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            boolean cancelled = future.cancel(true);
            if (!cancelled) {
                return false;
            }
            new ClientInvocation(client, JetCancelJobCodec.encodeRequest(executionId), partitionId)
                    .invoke().andThen(new ExecutionCallback<ClientMessage>() {
                @Override
                public void onResponse(ClientMessage clientMessage) {
                    //ignored
                }

                @Override
                public void onFailure(Throwable throwable) {
                    logger.warning("Error cancelling job with id " + executionId, throwable);
                }
            });
            return true;
        }

        @Override
        public boolean isCancelled() {
            return future.isCancelled();
        }

        @Override
        public boolean isDone() {
            return future.isDone();
        }

        @Override
        public Void get() throws InterruptedException, ExecutionException {
            future.get();
            return null;
        }

        @Override
        public Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
            future.get(timeout, unit);
            return null;
        }
    }
}

@@ -1273,24 +1273,19 @@
 jet.
-config.JetConfig
+JetInstance
 ;%0Aim
@@ -1304,34 +1304,26 @@
 elcast.jet.J
-etInstance
+ob
 ;%0Aimport com
@@ -1329,35 +1329,48 @@
 m.hazelcast.jet.
-Job
+config.JetConfig
 ;%0Aimport com.haz
@@ -1580,24 +1580,58 @@
 ng.ILogger;%0A
+import com.hazelcast.nio.Address;%0A
 import com.h
@@ -1704,24 +1704,25 @@
 n.Supplier;%0A
+%0A
 import java.
@@ -3314,31 +3314,40 @@
         
-int partitionId
+Address executionAddress
  = clien
@@ -3403,19 +3403,26 @@
 .get
-PartitionId
+Owner().getAddress
 ();%0A
@@ -3568,27 +3568,32 @@
 gData), 
-partitionId
+executionAddress
 );%0A     
@@ -3660,27 +3660,32 @@
 tionId, 
-partitionId
+executionAddress
 );%0A     
@@ -5306,31 +5306,40 @@
 e final 
-int partitionId
+Address executionAddress
 ;%0A%0A     
@@ -5420,23 +5420,32 @@
 Id, 
-int partitionId
+Address executionAddress
 ) %7B%0A
@@ -5543,33 +5543,43 @@
 his.
-partitionId = partitionId
+executionAddress = executionAddress
 ;%0A  
@@ -5886,19 +5886,24 @@
 d), 
-partitionId
+executionAddress
 )%0A  

Use Address instead of partitionId on client invocations (#281)