/**
 *  ServingXML
 *  
 *  Copyright (C) 2006  Daniel Parker
 *    daniel.parker@servingxml.com 
 * 
 * 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 com.servingxml.components.recordio.sql;

import java.sql.Connection;
import java.sql.Statement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.io.IOException;

import com.servingxml.components.recordio.RecordReader;
import com.servingxml.components.recordio.RecordWriter;
import com.servingxml.app.ServiceContext;
import com.servingxml.util.StringHelper;
import com.servingxml.util.Name;
import com.servingxml.util.QualifiedName;
import com.servingxml.util.ServingXmlException;
import com.servingxml.util.record.FieldType;
import com.servingxml.util.record.Record;
import com.servingxml.util.record.RecordBuilder;
import com.servingxml.util.record.RecordType;
import com.servingxml.components.connect.sql.JdbcConnectionPool;
import com.servingxml.io.cache.Expirable;
import com.servingxml.io.cache.Key;
import com.servingxml.io.cache.DefaultKey;
import com.servingxml.io.flow.Flow;

public class SqlReader implements RecordReader {

  private final JdbcConnectionPool connectionPool;
  private final SqlQuery sqlQuery;
  private final Key key;
  private RecordWriter recordWriter = null;

  public SqlReader(JdbcConnectionPool connectionPool, SqlQuery sqlQuery) 
  throws ServingXmlException {

    this.connectionPool = connectionPool;
    this.sqlQuery = sqlQuery;
    this.key = DefaultKey.newInstance();
  }

  protected void readRecords(ServiceContext context, Record parameters, Flow flow, ReadRecord readRecord) 
  throws ServingXmlException {

    Connection connection = null;
    ResultSet resultSet = null;
    Statement stat = null;
    boolean lastAutoCommit = false;

    String sql = sqlQuery.buildSql(context, parameters, flow);
    Name recordTypeName = sqlQuery.getRecordTypeName();
    
    try {
      connection = connectionPool.getConnection();
      lastAutoCommit = connection.getAutoCommit();
      connection.setAutoCommit(true);

      stat = connection.createStatement();
      resultSet = stat.executeQuery(sql);

      ResultSetMetaData metaData = resultSet.getMetaData();

      RecordBuilder builder = new RecordBuilder(recordTypeName);
      int columnCount = metaData.getColumnCount();
      for (int i = 1; i <= columnCount; ++i) {
        String columnName = metaData.getColumnName(i);
        builder.setField(new QualifiedName(columnName),"");
      }
      RecordType recordType = builder.toRecord().getRecordType();
      while (resultSet.next()) {
        RecordBuilder recordBuilder = new RecordBuilder(recordType);
        int count = recordType.count();
        for (int i = 0; i < count; ++i) {
          FieldType fieldType = recordType.getFieldTypeAt(i);
          String value = null;
          try {
            value = resultSet.getString(fieldType.getName().getLocalName());
            // Replace SQL NULL with empty string  
            if (value == null) {
              value = "";
            }
            if (sqlQuery.isTrimLeading() && sqlQuery.isTrimTrailing()) {
              value = value.trim();
            } else if (sqlQuery.isTrimLeading()) {
              value = StringHelper.trimLeading(value);
            } else if (sqlQuery.isTrimTrailing()) {
              value = StringHelper.trimTrailing(value);
            }
          } catch (SQLException e) {
          }
          if (value != null) {
            recordBuilder.setField(fieldType.getName(),value);
          }
        }
        Record record = recordBuilder.toRecord();
        readRecord.recordRead(record);
      }

    } catch (SQLException e) {
      throw new ServingXmlException(e.getMessage(),e);
    } finally {
      if (resultSet != null) {
        try {
          resultSet.close();
        } catch (Exception rse) {
          //  Don't care
        }
      }
      if (stat != null) {
        try {
          stat.close();
        } catch (Exception se) {
          //  Don't care
        }
      }
      if (connection != null) {
        try {
          connection.setAutoCommit(lastAutoCommit);
        } catch (SQLException e2) {
          throw new ServingXmlException(e2.getMessage());
        }
        connectionPool.releaseConnection(connection);
      }
    }
  }

  public Key getKey() throws ServingXmlException {
    return key;         
  }

  public Expirable getExpirable() throws ServingXmlException {
    return Expirable.ALWAYS_EXPIRED;
  }

  public void readRecords(final ServiceContext context, final Record parameters, final Flow flow) 
  throws ServingXmlException {
    if (recordWriter != null) {
      recordWriter.startRecordStream(context, parameters, flow);

      ReadRecord readRecord = new ReadRecord() {
        public void recordRead(Record record) throws ServingXmlException {
          //try {
            Flow newFlow = flow.modifyCurrentRecord(context, record);
            recordWriter.writeRecord(context, parameters,newFlow);
          //} catch (ServingXmlException e) {
          //  context.error(e.getMessage());
          //}
        }
      };
      readRecords(context, parameters, flow, readRecord);

      recordWriter.endRecordStream(context, parameters, flow);
    } 
  }

  public void setRecordWriter(RecordWriter recordWriter) {
    this.recordWriter = recordWriter;
  }

  public void close() throws IOException {
  }
}

interface ReadRecord {
  void recordRead(Record record) throws ServingXmlException;
}



