// Path: src/main/java/crate/elasticsearch/action/searchinto/SearchIntoContext.java
// public class SearchIntoContext extends SearchContext {
// 
//     // currently we only support index targets
//     private String targetType = "index";
// 
//     private List<InetSocketTransportAddress> targetNodes;
// 
// 
//     public Map<String, String> outputNames() {
//         return outputNames;
//     }
// 
//     private final Map<String, String> outputNames = new HashMap<String,
//             String>();
// 
//     public SearchIntoContext(long id, ShardSearchRequest request,
//             SearchShardTarget shardTarget, Engine.Searcher engineSearcher,
//             IndexService indexService, IndexShard indexShard,
//             ScriptService scriptService, CacheRecycler cacheRecycler) {
//         super(id, request, shardTarget, engineSearcher, indexService,
//                 indexShard, scriptService, cacheRecycler);
//     }
// 
//     public String targetType() {
//         // this is currently the only type supported
//         return targetType;
//     }
// 
//     public List<InetSocketTransportAddress> targetNodes() {
//         if (targetNodes == null) {
//             targetNodes = Lists.newArrayList();
//         }
//         return targetNodes;
//     }
// 
//     public void emptyTargetNodes() {
//         this.targetNodes = ImmutableList.of();
//     }
// 
// }


import crate.elasticsearch.action.searchinto.SearchIntoContext;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.assistedinject.Assisted;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.Loggers;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.BaseFuture;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.fetch.FetchSubPhase;
import org.elasticsearch.search.fetch.script.ScriptFieldsFetchSubPhase;
import org.elasticsearch.search.fetch.version.VersionFetchSubPhase;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
package crate.elasticsearch.searchinto;



public class BulkWriterCollector extends WriterCollector {


    private static final ESLogger logger = Loggers.getLogger(
            BulkWriterCollector.class);

    private Client client;
    private Client transportClient;
    private BulkProcessor bulkProcessor;

    private final AtomicLong startedDocs = new AtomicLong(0);
    private final AtomicLong succeededDocs = new AtomicLong(0);

    public static final String NAME = "index";
    private BulkListener bulkListener;

    public BulkWriterCollector() {
        super();
    }

    @Inject
    public BulkWriterCollector(@Assisted SearchIntoContext context,
            Client client, ScriptFieldsFetchSubPhase scriptFieldsPhase,
            VersionFetchSubPhase versionFetchSubPhase) {
        super(context,
                new FetchSubPhase[]{versionFetchSubPhase, scriptFieldsPhase});
        this.client = client;
    }

    /**
     * determine which client to use depending on context
     *
     * @return injected client or TransportClient
     */
    private Client getClient() {

// Path: src/main/java/crate/elasticsearch/action/searchinto/SearchIntoContext.java
// public class SearchIntoContext extends SearchContext {
// 
//     // currently we only support index targets
//     private String targetType = "index";
// 
//     private List<InetSocketTransportAddress> targetNodes;
// 
// 
//     public Map<String, String> outputNames() {
//         return outputNames;
//     }
// 
//     private final Map<String, String> outputNames = new HashMap<String,
//             String>();
// 
//     public SearchIntoContext(long id, ShardSearchRequest request,
//             SearchShardTarget shardTarget, Engine.Searcher engineSearcher,
//             IndexService indexService, IndexShard indexShard,
//             ScriptService scriptService, CacheRecycler cacheRecycler) {
//         super(id, request, shardTarget, engineSearcher, indexService,
//                 indexShard, scriptService, cacheRecycler);
//     }
// 
//     public String targetType() {
//         // this is currently the only type supported
//         return targetType;
//     }
// 
//     public List<InetSocketTransportAddress> targetNodes() {
//         if (targetNodes == null) {
//             targetNodes = Lists.newArrayList();
//         }
//         return targetNodes;
//     }
// 
//     public void emptyTargetNodes() {
//         this.targetNodes = ImmutableList.of();
//     }
// 
// }


// Path: src/main/java/crate/elasticsearch/searchinto/BulkWriterCollector.java
import crate.elasticsearch.action.searchinto.SearchIntoContext;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.inject.assistedinject.Assisted;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.Loggers;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.BaseFuture;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.fetch.FetchSubPhase;
import org.elasticsearch.search.fetch.script.ScriptFieldsFetchSubPhase;
import org.elasticsearch.search.fetch.version.VersionFetchSubPhase;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;

package crate.elasticsearch.searchinto;



public class BulkWriterCollector extends WriterCollector {


    private static final ESLogger logger = Loggers.getLogger(
            BulkWriterCollector.class);

    private Client client;
    private Client transportClient;
    private BulkProcessor bulkProcessor;

    private final AtomicLong startedDocs = new AtomicLong(0);
    private final AtomicLong succeededDocs = new AtomicLong(0);

    public static final String NAME = "index";
    private BulkListener bulkListener;

    public BulkWriterCollector() {
        super();
    }

    @Inject
    public BulkWriterCollector(@Assisted SearchIntoContext context,
            Client client, ScriptFieldsFetchSubPhase scriptFieldsPhase,
            VersionFetchSubPhase versionFetchSubPhase) {
        super(context,
                new FetchSubPhase[]{versionFetchSubPhase, scriptFieldsPhase});
        this.client = client;
    }

    /**
     * determine which client to use depending on context
     *
     * @return injected client or TransportClient
     */
    private Client getClient() {

        SearchIntoContext ctx = (SearchIntoContext)context;