/**
 * Copyright (C) 2014 Stratio (http://stratio.com)
 * <p/>
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.stratio.streaming.functions;

import com.stratio.streaming.commons.messages.ColumnNameTypeValue;
import com.stratio.streaming.commons.messages.StratioStreamingMessage;

import com.stratio.streaming.service.SolrOperationsService;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.common.SolrInputDocument;
import org.hsqldb.lib.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.xml.sax.SAXException;

import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.TransformerException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;
import java.util.Collection;
import java.util.HashMap;

public class SaveToSolrActionExecutionFunction extends BaseActionExecutionFunction {

    private static final long serialVersionUID = 3522740757019463301L;

    private static final Logger log = LoggerFactory.getLogger(SaveToSolrActionExecutionFunction.class);

    private Map<String, SolrClient> solrClients = new HashMap<>();
    private List<String> solrCores = new ArrayList<String>();

    private SolrOperationsService solrOperationsService;

    private final String dataDir;
    private final String solrHosts;
    private final Boolean isCloud;

    public SaveToSolrActionExecutionFunction(String solrHosts, Boolean isCloud, String dataDir) {
        this.solrHosts = solrHosts;
        this.dataDir = dataDir;
        this.isCloud = isCloud;
        this.solrOperationsService = new SolrOperationsService(solrHosts, dataDir, isCloud);
    }

    @Override
    public void process(Iterable<StratioStreamingMessage> messages) throws Exception {
        Map<String, Collection<SolrInputDocument>> elemntsToInsert = new HashMap<String, Collection<SolrInputDocument>>();
        for (StratioStreamingMessage stratioStreamingMessage : messages) {
            SolrInputDocument document = new SolrInputDocument();
            document.addField("id", stratioStreamingMessage.getTimestamp());
            for (ColumnNameTypeValue column : stratioStreamingMessage.getColumns()) {
                document.addField(column.getColumn(), column.getValue());
            }
            Collection<SolrInputDocument> collection = elemntsToInsert.get(stratioStreamingMessage.getStreamName());
            collection.add(document);
            elemntsToInsert.put(stratioStreamingMessage.getStreamName(), collection);
            getClient(stratioStreamingMessage).add(document);
        }
        for (Map.Entry<String, Collection<SolrInputDocument>> elem: elemntsToInsert.entrySet()) {
            getSolrclient(elem.getKey()).add(elem.getValue());
        }
        flushClients();
    }

    private SolrClient getClient(StratioStreamingMessage message) throws IOException, SolrServerException, URISyntaxException, TransformerException, SAXException, ParserConfigurationException {
        String core = message.getStreamName();
        if (solrClients.containsKey(core)) {
            //we have a client for this core
            return solrClients.get(core);
        } else {
            //check if core exists
            if (solrCores.size() == 0) {
                // Initialize solrcores list
                solrCores = solrOperationsService.getCoreList();
            }
            if (!solrCores.contains(core)) {
                // Create Core
                solrOperationsService.createCore(message);
                // Update core list
                solrCores = solrOperationsService.getCoreList();
            }
            SolrClient solrClient = getSolrclient(core);
            solrClients.put(core, solrClient);
            return solrClient;
        }
    }

    private void flushClients() throws IOException, SolrServerException, URISyntaxException {
        //Do commit in all Solrclients
        for (String core : solrClients.keySet()) {
            getSolrclient(core).commit();
        }
    }

    private SolrClient getSolrclient(String core) {
        SolrClient solrClient;
        if (isCloud) {
            solrClient = new CloudSolrClient(solrHosts);
            ((CloudSolrClient)solrClient).setDefaultCollection(core);
        } else {
            solrClient = new HttpSolrClient("http://" + solrHosts + "/solr/" + core);
        }
        return solrClient;
    }



}

@@ -3017,32 +3017,80 @@
 ;%0A            %7D%0A
+            checkCore(stratioStreamingMessage);%0A
             Coll
@@ -3604,28 +3604,22 @@
 ate 
-SolrClient getClient
+void checkCore
 (Str
@@ -3685,33 +3685,43 @@
 rException, 
-URISyntax
+ParserConfiguration
 Exception, T
@@ -3755,35 +3755,25 @@
 eption, 
-ParserConfiguration
+URISyntax
 Exceptio
@@ -3827,161 +3827,8 @@
 ();%0A
-        if (solrClients.containsKey(core)) %7B%0A            //we have a client for this core%0A            return solrClients.get(core);%0A        %7D else %7B%0A    
     
@@ -3858,28 +3858,24 @@
 sts%0A        
-  
-  
 if (solrCore
@@ -3887,28 +3887,24 @@
 e() == 0) %7B%0A
-    
             
@@ -3928,28 +3928,24 @@
 rcores list%0A
-    
             
@@ -3997,34 +3997,26 @@
 ();%0A        
-    %7D%0A    
+%7D%0A
         if (
@@ -4056,20 +4056,16 @@
         
-  
-  
 // Creat
@@ -4071,20 +4071,16 @@
 te Core%0A
-    
         
@@ -4138,20 +4138,16 @@
         
-    
 // Updat
@@ -4162,36 +4162,32 @@
 ist%0A            
- 
-   
 solrCores = solr
@@ -4219,37 +4219,430 @@
 List();%0A        
+%7D%0A
     %7D
+%0A%0A    private SolrClient getClient(StratioStreamingMessage message) throws IOException, SolrServerException, URISyntaxException, TransformerException, SAXException, ParserConfigurationException %7B%0A        String core = message.getStreamName();%0A        if (solrClients.containsKey(core)) %7B%0A            //we have a client for this core%0A            return solrClients.get(core);%0A        %7D else %7B
 %0A            Sol

Improve reading to solr