/*
 * Copyright 2014-2015 Red Hat, Inc. and/or its affiliates
 * and other contributors as indicated by the @author tags.
 *
 * 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 org.hawkular.metrics.core.impl.schema;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.StringWriter;
import java.util.Map;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Session;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.CharStreams;
import com.google.common.io.Closeables;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.hawkular.metrics.core.impl.util.TokenReplacingReader;

/**
 * @author John Sanda
 * @author Heiko W. Rupp
 */
public class SchemaManager {

    private static final Logger logger = LoggerFactory.getLogger(SchemaManager.class);

    private Session session;

    public SchemaManager(Session session) {
        this.session = session;
    }

    public void dropKeyspace(String keyspace) {
        logger.info("Dropping keyspace " + keyspace);
        session.execute("DROP KEYSPACE " + keyspace);
    }

    public void createSchema(String keyspace) throws IOException {
        logger.info("Creating schema for keyspace " + keyspace);

        ResultSet resultSet = session.execute("SELECT * FROM system.schema_keyspaces WHERE keyspace_name = '" +
            keyspace + "'");
        if (!resultSet.isExhausted()) {
            logger.info("Schema already exist. Skipping schema creation.");
            return;
        }

        ImmutableMap<String, String> schemaVars = ImmutableMap.of("keyspace", keyspace);

        InputStream inputStream = getClass().getResourceAsStream("/schema.cql");
        InputStreamReader reader = new InputStreamReader(inputStream);
        String content = CharStreams.toString(reader);

        for (String cql : content.split("(?m)^-- #.*$")) {
            if (!cql.startsWith("--")) {
                TokenReplacingReader tokenReader = new TokenReplacingReader(cql.trim(), schemaVars);
                String updatedCQL = substituteVars(cql.trim(), schemaVars);
                logger.info("Executing CQL:\n" + updatedCQL + "\n");
                session.execute(updatedCQL);
            }
        }
    }

    private String substituteVars(String cql, Map<String, String> vars) {
        TokenReplacingReader reader = new TokenReplacingReader(cql, vars);
        StringWriter writer = new StringWriter();
        try {
            char[] buffer = new char[32768];
            int cnt;
            while ((cnt = reader.read(buffer)) != -1) {
                writer.write(buffer, 0, cnt);
            }
            return writer.toString();
        } catch (IOException e) {
            throw new RuntimeException("Failed to perform variable substition on CQL", e);
        } finally {
            try {
                Closeables.close(reader, true);
                Closeables.close(writer, true);
            } catch (IOException e) {
                logger.info("There was a problem closing resources", e);
            }
        }
    }

}


@@ -1042,105 +1042,8 @@
 ms;%0A
-import com.google.common.io.Closeables;%0A%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0Aimp
@@ -1103,16 +1103,73 @@
 gReader;
+%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;
 %0A%0A/**%0A *
@@ -2122,32 +2122,37 @@
 pace);%0A%0A        
+try (
 InputStream inpu
@@ -2216,16 +2216,20 @@
         
+    
 InputStr
@@ -2281,26 +2281,29 @@
 tStream)
-;%0A
+) %7B%0A 
         
 String c
@@ -2290,24 +2290,27 @@
  %7B%0A         
+   
 String conte
@@ -2341,24 +2341,28 @@
 g(reader);%0A%0A
+    
         for 
@@ -2416,24 +2416,28 @@
             
+    
 if (!cql.sta
@@ -2461,105 +2461,8 @@
     
-            TokenReplacingReader tokenReader = new TokenReplacingReader(cql.trim(), schemaVars);%0A
     
@@ -2541,32 +2541,36 @@
                 
+    
 logger.info(%22Exe
@@ -2614,32 +2614,36 @@
                 
+    
 session.execute(
@@ -2655,16 +2655,34 @@
 edCQL);%0A
+                %7D%0A
         
@@ -2786,16 +2786,21 @@
         
+try (
 TokenRep
@@ -2862,24 +2862,28 @@
 s);%0A        
+    
 StringWriter
@@ -2914,21 +2914,9 @@
 er()
-;%0A        try
+)
  %7B%0A 
@@ -3263,267 +3263,8 @@
 e);%0A
-        %7D finally %7B%0A            try %7B%0A                Closeables.close(reader, true);%0A                Closeables.close(writer, true);%0A            %7D catch (IOException e) %7B%0A                logger.info(%22There was a problem closing resources%22, e);%0A            %7D%0A
     

Use try-with for closeable resources and remove unused tokenReader (resource leak possibility)