package com.senseidb.federated.broker;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.json.JSONObject;

import com.senseidb.federated.broker.proxy.BrokerProxy;
import com.senseidb.federated.broker.proxy.SenseiBrokerProxy;
import com.senseidb.search.node.Broker;
import com.senseidb.search.node.ResultMerger;
import com.senseidb.search.node.SenseiBroker;
import com.senseidb.search.node.inmemory.InMemorySenseiService;
import com.senseidb.search.req.SenseiRequest;
import com.senseidb.search.req.SenseiResult;
import com.senseidb.search.req.SenseiSystemInfo;
import com.senseidb.servlet.AbstractSenseiClientServlet;
import com.senseidb.servlet.DefaultSenseiJSONServlet;
import com.senseidb.svc.api.SenseiException;
import com.senseidb.util.RequestConverter2;

public class FederatedBroker implements Broker<SenseiRequest, SenseiResult>{
  private final static Logger logger = Logger.getLogger(SenseiBrokerProxy.class);
  private List<BrokerProxy> proxies;
  private int numThreads = 10;  
  private ExecutorService executor;
  private long timeout = 8000;
  
  private Map<String, String[]> facetInfo = new HashMap<String, String[]>();
  public FederatedBroker() {
  }
  public FederatedBroker(List<BrokerProxy> proxies) {
    this.proxies = proxies;
  }
  public void start() {
    executor = Executors.newFixedThreadPool(numThreads);
  }
  
  public List<BrokerProxy> getProxies() {
    return proxies;
  }
  public void setProxies(List<BrokerProxy> proxies) {
    this.proxies = proxies;
  }
  
  public int getNumThreads() {
    return numThreads;
  }
  public void setNumThreads(int numThreads) {
    this.numThreads = numThreads;
  }
  
  
  public long getTimeout() {
    return timeout;
  }
  public void setTimeout(long timeout) {
    this.timeout = timeout;
  }
  @Override
  public SenseiResult browse(final SenseiRequest request) throws SenseiException {
    final List<SenseiResult> resultList = Collections.synchronizedList(new ArrayList<SenseiResult>());
    final CountDownLatch countDownLatch = new CountDownLatch(proxies.size());
    for (final BrokerProxy proxy : proxies) {
      executor.submit(new Runnable() {        
        public void run() {
          try {
            resultList.addAll(proxy.doQuery(request));
            countDownLatch.countDown();
          } catch (Exception ex) {
            logger.error("Error while calling the proxy", ex);
          }
        }
      });      
    }
    try {
      boolean allTheResults = countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
      if (!allTheResults) {        
        logger.warn("Not all the results are received");
      }
      SenseiResult res = ResultMerger.merge(request, resultList, false);      
      if (request.isFetchStoredFields() || request.isFetchStoredValue())
        SenseiBroker.recoverSrcData(res, res.getSenseiHits(), request.isFetchStoredFields());
      return res;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }  
  }
  
 
  public void setInMemorySenseiService(InMemorySenseiService inMemorySenseiService) {
    if (inMemorySenseiService != null && inMemorySenseiService.getSenseiSystemInfo() != null) {
      facetInfo = AbstractSenseiClientServlet.extractFacetInfo(inMemorySenseiService.getSenseiSystemInfo());      
    }
  }
  public JSONObject query(JSONObject request) {
    try {
      SenseiRequest senseiRequest = RequestConverter2.fromJSON(request, facetInfo);
      SenseiResult senseiResult = browse(senseiRequest);
      JSONObject jsonResult = DefaultSenseiJSONServlet.buildJSONResult(senseiRequest, senseiResult);
      return jsonResult;
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  } 
  public void stop() {
    executor.shutdown();
  }
}

@@ -1045,16 +1045,185 @@
 rter2;%0A%0A
+import static com.senseidb.servlet.SenseiSearchServletParams.PARAM_RESULT_NUMHITS;%0Aimport static com.senseidb.servlet.SenseiSearchServletParams.PARAM_RESULT_TOTALDOCS;%0A%0A
 public c
@@ -1373,16 +1373,101 @@
 class);%0A
+  private final static Logger queryLogger = Logger.getLogger(%22com.sensei.querylog%22);%0A
   privat
@@ -3863,24 +3863,122 @@
 %7B%0A    try %7B%0A
+      long time = System.currentTimeMillis();%0A      int numHits = 0, totalDocs = 0;%0A      try %7B%0A  
       Sensei
@@ -4051,24 +4051,26 @@
 nfo);%0A      
+ 
+ 
 SenseiResult
@@ -4104,24 +4104,26 @@
 eiRequest);%0A
+  
       JSONOb
@@ -4221,26 +4221,449 @@
     
-return jsonResult;
+  if (jsonResult != null) %7B%0A          numHits = jsonResult.optInt(PARAM_RESULT_NUMHITS);%0A          totalDocs = jsonResult.optInt(PARAM_RESULT_TOTALDOCS);%0A        %7D%0A        return jsonResult;%0A      %7D%0A      finally %7B%0A        if (queryLogger.isInfoEnabled() && request != null) %7B%0A          queryLogger.info(String.format(%22hits(%25d/%25d) took %25dms: %25s%22, numHits, totalDocs, System.currentTimeMillis() - time, request.toString()));%0A        %7D%0A      %7D
 %0A   

add query log for federated broker.