/*
 * Licensed to Crate under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.  Crate licenses this file
 * to you 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.
 *
 * However, if you have executed another commercial license agreement
 * with Crate these terms will supersede the license and you may use the
 * software solely pursuant to the terms of the relevant commercial
 * agreement.
 */

package io.crate.operation.projectors.fetch;

import com.carrotsearch.hppc.IntContainer;
import com.carrotsearch.hppc.IntObjectMap;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
import io.crate.Streamer;
import io.crate.core.collections.Bucket;
import io.crate.executor.transport.NodeFetchRequest;
import io.crate.executor.transport.NodeFetchResponse;
import io.crate.executor.transport.TransportFetchNodeAction;
import org.elasticsearch.action.ActionListener;

import java.util.Map;
import java.util.UUID;

public class TransportFetchOperation implements FetchOperation {

    private final TransportFetchNodeAction transportFetchNodeAction;
    private final Map<String, ? extends IntObjectMap<Streamer[]>> nodeIdToReaderIdToStreamers;
    private final UUID jobId;
    private final int executionPhaseId;

    public TransportFetchOperation(TransportFetchNodeAction transportFetchNodeAction,
                                   Map<String, ? extends IntObjectMap<Streamer[]>> nodeIdToReaderIdToStreamers,
                                   UUID jobId,
                                   int executionPhaseId) {
        this.transportFetchNodeAction = transportFetchNodeAction;
        this.nodeIdToReaderIdToStreamers = nodeIdToReaderIdToStreamers;
        this.jobId = jobId;
        this.executionPhaseId = executionPhaseId;
    }

    @Override
    public ListenableFuture<IntObjectMap<? extends Bucket>> fetch(String nodeId,
                                                                  IntObjectMap<? extends IntContainer> toFetch,
                                                                  boolean closeContext) {
        final SettableFuture<IntObjectMap<? extends Bucket>> future = SettableFuture.create();
        transportFetchNodeAction.execute(
            nodeId,
            nodeIdToReaderIdToStreamers.get(nodeId),
            new NodeFetchRequest(jobId, executionPhaseId, closeContext, toFetch),
            new ActionListener<NodeFetchResponse>() {
                @Override
                public void onResponse(NodeFetchResponse nodeFetchResponse) {
                    future.set(nodeFetchResponse.fetched());
                }

                @Override
                public void onFailure(Throwable e) {
                    future.setException(e);
                }
            });
        return future;
    }
}

@@ -1153,40 +1153,21 @@
 mon.
-util.concurrent.ListenableFuture
+base.Function
 ;%0Aim
@@ -1205,20 +1205,22 @@
 current.
-Sett
+Listen
 ableFutu
@@ -1245,24 +1245,69 @@
 e.Streamer;%0A
+import io.crate.action.FutureActionListener;%0A
 import io.cr
@@ -1507,56 +1507,42 @@
 on;%0A
+%0A
 import 
-org.elasticsearch.action.ActionListener;%0A
+javax.annotation.Nullable;
 %0Aimp
@@ -1650,16 +1650,393 @@
 tion %7B%0A%0A
+    private static final Function%3CNodeFetchResponse, IntObjectMap%3C? extends Bucket%3E%3E GET_FETCHED =%0A        new Function%3CNodeFetchResponse, IntObjectMap%3C? extends Bucket%3E%3E() %7B%0A%0A        @Nullable%0A        @Override%0A        public IntObjectMap%3C? extends Bucket%3E apply(@Nullable NodeFetchResponse input) %7B%0A            return input == null ? null : input.fetched();%0A        %7D%0A    %7D;%0A
     priv
@@ -3098,92 +3098,129 @@
     
-final SettableFuture%3CIntObjectMap%3C? extends Bucket%3E%3E future = SettableFuture.create(
+FutureActionListener%3CNodeFetchResponse, IntObjectMap%3C? extends Bucket%3E%3E listener = new FutureActionListener%3C%3E(GET_FETCHED
 );%0A 
@@ -3431,412 +3431,42 @@
     
-new ActionListener%3CNodeFetchResponse%3E() %7B%0A                @Override%0A                public void onResponse(NodeFetchResponse nodeFetchResponse) %7B%0A                    future.set(nodeFetchResponse.fetched());%0A                %7D%0A%0A                @Override%0A                public void onFailure(Throwable e) %7B%0A                    future.setException(e);%0A                %7D%0A            %7D);%0A        return future
+listener);%0A        return listener
 ;%0A  

use FutureActionListener in TransportFetchOperation