Add RDBMS table iterator implementation
package org.wso2.siddhi.extension.table.rdbms;
import org.wso2.siddhi.core.table.record.RecordIterator;
+import org.wso2.siddhi.extension.table.rdbms.exception.RDBMSTableException;
import org.wso2.siddhi.extension.table.rdbms.util.RDBMSTableUtils;
+import org.wso2.siddhi.query.api.definition.Attribute;
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
public class RDBMSIterator implements RecordIterator<Object[]> {
- private ResultSet rs;
- private PreparedStatement stmt;
private Connection conn;
+ private PreparedStatement stmt = null;
+ private ResultSet rs = null;
+
+ private boolean preFetched;
+ private Object[] nextValue;
+ private List<Attribute> attributes;
+ private String tableName;
+
+ public RDBMSIterator(Connection conn, PreparedStatement stmt, ResultSet rs, List<Attribute> attributes, String tableName) {
+ this.conn = conn;
+ this.stmt = stmt;
+ this.rs = rs;
+ this.attributes = attributes;
+ this.tableName = tableName;
+ }
@Override
public boolean hasNext() {
- return false;
+ if (!this.preFetched) {
+ this.nextValue = this.next();
+ this.preFetched = true;
+ }
+ return nextValue != null;
}
@Override
public Object[] next() {
- return new Object[0];
+ if (this.preFetched) {
+ this.preFetched = false;
+ Object[] result = this.nextValue;
+ this.nextValue = null;
+ return result;
+ }
+ try {
+ if (this.rs.next()) {
+ return this.extractRecord(this.rs);
+ } else {
+ // end of the result set, cleaning up.
+ RDBMSTableUtils.cleanupConnection(this.rs, this.stmt, this.conn);
+ this.rs = null;
+ this.stmt = null;
+ this.conn = null;
+ return null;
+ }
+ } catch (Exception e) {
+ RDBMSTableUtils.cleanupConnection(this.rs, this.stmt, this.conn);
+ throw new RDBMSTableException("Error retrieving records from table '" + this.tableName + "': "
+ + e.getMessage(), e);
+ }
+ }
+
+ private Object[] extractRecord(ResultSet rs) throws SQLException {
+ int ordinal = 1;
+ List<Object> result = new ArrayList<>();
+ for (Attribute attribute : this.attributes) {
+ switch (attribute.getType()) {
+ case BOOL:
+ result.add(rs.getBoolean(ordinal));
+ break;
+ case DOUBLE:
+ result.add(rs.getDouble(ordinal));
+ break;
+ case FLOAT:
+ result.add(rs.getFloat(ordinal));
+ break;
+ case INT:
+ result.add(rs.getInt(ordinal));
+ break;
+ case LONG:
+ result.add(rs.getLong(ordinal));
+ break;
+ case OBJECT:
+ result.add(rs.getObject(ordinal));
+ break;
+ case STRING:
+ result.add(rs.getString(ordinal));
+ break;
+ }
+ ordinal++;
+ }
+ return result.toArray();
}
@Override
public void remove() {
-
+ //Do nothing. This is a read-only iterator.
}
@Override
