modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/rest/MetricHandlerBean.java | 111 +++----- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateMetricMapper.java | 134 ++++++---- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java | 75 +---- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexResultSetMapper.java | 33 ++ modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/PagedResultSet.java | 118 ++++++++ modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Query.java | 4 modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawNumericMetricMapper.java | 106 +++++-- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ResultSetMapper.java | 37 ++ 8 files changed, 408 insertions(+), 210 deletions(-)
New commits: commit cfb0a69d4bb0ff29fc0917388f72fa8a37a8e239 Author: Stefan Negrea snegrea@redhat.com Date: Fri Jan 25 12:24:02 2013 -0600
Update REST metric handler bean to get results for streaming raw metrics from Cassandra.
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/rest/MetricHandlerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/rest/MetricHandlerBean.java index 8271f46..05b89e9 100644 --- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/rest/MetricHandlerBean.java +++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/rest/MetricHandlerBean.java @@ -22,10 +22,6 @@ import java.io.IOException; import java.io.OutputStream; import java.io.PrintWriter; import java.net.URI; -import java.sql.Connection; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; import java.util.ArrayList; import java.util.Collection; import java.util.Date; @@ -83,13 +79,12 @@ import org.rhq.core.domain.measurement.composite.MeasurementDataNumericHighLowCo import org.rhq.core.domain.resource.Resource; import org.rhq.core.domain.resource.ResourceType; import org.rhq.core.domain.resource.group.ResourceGroup; -import org.rhq.core.util.jdbc.JDBCUtil; import org.rhq.enterprise.server.RHQConstants; +import org.rhq.enterprise.server.cassandra.SessionManagerBean; import org.rhq.enterprise.server.measurement.MeasurementAggregate; import org.rhq.enterprise.server.measurement.MeasurementDataManagerLocal; import org.rhq.enterprise.server.measurement.MeasurementDefinitionManagerLocal; import org.rhq.enterprise.server.measurement.MeasurementScheduleManagerLocal; -import org.rhq.enterprise.server.measurement.util.MeasurementDataManagerUtility; import org.rhq.enterprise.server.resource.ResourceManagerLocal; import org.rhq.enterprise.server.resource.group.ResourceGroupManagerLocal; import org.rhq.enterprise.server.rest.domain.Baseline; @@ -99,6 +94,9 @@ import org.rhq.enterprise.server.rest.domain.MetricAggregate; import org.rhq.enterprise.server.rest.domain.MetricSchedule; import org.rhq.enterprise.server.rest.domain.NumericDataPoint; import org.rhq.enterprise.server.rest.domain.StringValue; +import org.rhq.server.metrics.MetricsDAO; +import org.rhq.server.metrics.PagedResultSet; +import org.rhq.server.metrics.RawNumericMetric;
/** * Deal with metrics @@ -126,6 +124,10 @@ public class MetricHandlerBean extends AbstractRestBean { ResourceManagerLocal resMgr; @EJB ResourceGroupManagerLocal groupMgr; + + @EJB + private SessionManagerBean sessionManager; + @PersistenceContext(unitName = RHQConstants.PERSISTENCE_UNIT_NAME) EntityManager em;
@@ -636,7 +638,6 @@ public class MetricHandlerBean extends AbstractRestBean { so.startTime = startTime; so.endTime = endTime; so.mediaType = mediaType; - so.now = now-1; // pass this so that the for the 7days case is still handled from raw tables.
return so; } @@ -844,45 +845,20 @@ public class MetricHandlerBean extends AbstractRestBean { int scheduleId; long startTime; long endTime; - long now; MediaType mediaType;
@Override public void write(OutputStream outputStream) throws IOException, WebApplicationException { + MetricsDAO metricsDAO = new MetricsDAO(sessionManager.getSession()); + PagedResultSet<RawNumericMetric> resultSet = metricsDAO.findRawMetricsPaged(scheduleId, startTime, endTime);
- String[] tables = MeasurementDataManagerUtility.getTables(startTime,endTime,now); - StringBuilder sb = new StringBuilder(); - for (int i = 0 ; i < tables.length ; i ++) { - sb.append("SELECT time_stamp,value FROM "); - sb.append(tables[i]); - sb.append(" WHERE schedule_id = ? AND time_stamp BETWEEN ? AND ?"); - if (i < tables.length-1) - sb.append(" UNION ALL "); - } - - sb.append(" ORDER BY time_stamp ASC"); - + PrintWriter pw = new PrintWriter(outputStream);
- - Connection connection = null; - PreparedStatement ps = null; - ResultSet rs = null; - try { - connection = rhqDs.getConnection(); - ps = connection.prepareStatement( sb.toString() ); - for (int i = 0; i < tables.length ; i++) { - ps.setInt(i * 3 + 1, scheduleId); - ps.setLong(i*3+2,startTime); - ps.setLong(i*3+3,endTime); - } - rs = ps.executeQuery(); - - PrintWriter pw = new PrintWriter(outputStream); - - if (mediaType.equals(MediaType.APPLICATION_JSON_TYPE)) { - boolean needsComma = false; - pw.println("["); - while (rs.next()) { + if (mediaType.equals(MediaType.APPLICATION_JSON_TYPE)) { + boolean needsComma = false; + pw.println("["); + while (!resultSet.isExhausted()) { + for (RawNumericMetric metric : resultSet.getNextPage()) { if (needsComma) { pw.print(",\n"); } @@ -892,59 +868,58 @@ public class MetricHandlerBean extends AbstractRestBean { pw.print(scheduleId); pw.print(", "); pw.print(""timeStamp":"); - pw.print(rs.getLong(1)); + pw.print(metric.getTimestamp()); pw.print(", "); pw.print(""value":"); - pw.print(rs.getDouble(2)); + pw.print(metric.getValue()); pw.print("}"); } - pw.println("]"); } - else if (mediaType.equals(MediaType.APPLICATION_XML_TYPE)) { - pw.println("<collection>"); - while(rs.next()) { + pw.println("]"); + } else if (mediaType.equals(MediaType.APPLICATION_XML_TYPE)) { + pw.println("<collection>"); + while (!resultSet.isExhausted()) { + for (RawNumericMetric metric : resultSet.getNextPage()) { pw.print(" <numericDataPoint scheduleId=""); pw.print(scheduleId); pw.print("" timeStamp=""); - pw.print(rs.getLong(1)); + pw.print(metric.getTimestamp()); pw.print("" value=""); - pw.print(rs.getDouble(2)); + pw.print(metric.getValue()); pw.println(""/>"); } - pw.println("</collection>"); } - else if (mediaType.toString().equals("text/csv")) { - pw.println("#schedule,timestamp,value"); - while (rs.next()) { + pw.println("</collection>"); + } else if (mediaType.toString().equals("text/csv")) { + pw.println("#schedule,timestamp,value"); + while (!resultSet.isExhausted()) { + for (RawNumericMetric metric : resultSet.getNextPage()) { pw.print(scheduleId); pw.print(','); - pw.print(rs.getLong(1)); + pw.print(metric.getTimestamp()); pw.print(','); - pw.println(rs.getDouble(2)); + pw.println(metric.getValue()); } } - else if (mediaType.equals(MediaType.TEXT_HTML_TYPE)) { - pw.println("<table>"); - pw.print("<tr><th>time</th><th>value</th></tr>\n"); - while (rs.next()) { + } else if (mediaType.equals(MediaType.TEXT_HTML_TYPE)) { + pw.println("<table>"); + pw.print("<tr><th>time</th><th>value</th></tr>\n"); + while (!resultSet.isExhausted()) { + for (RawNumericMetric metric : resultSet.getNextPage()) { pw.print(" <tr>"); pw.print("<td>"); - pw.print(new Date(rs.getLong(1))); + pw.print(new Date(metric.getTimestamp())); pw.print("</td><td>"); - pw.print(rs.getDouble(2)); + pw.print(metric.getValue()); pw.print("</td>"); pw.println("</tr>"); } - pw.println("</table>"); - } - pw.flush(); - pw.close(); - } catch (SQLException e) { - log.error(e); - } finally { - JDBCUtil.safeClose(connection, ps, rs); + pw.println("</table>"); + } + pw.flush(); + pw.close(); } } }
commit 78e5e519f0fb069fffc803465aa4f98cf06945a5 Author: Stefan Negrea snegrea@redhat.com Date: Fri Jan 25 12:23:21 2013 -0600
Add paged result sets mapped to domain objects to the metrics DAO interface. So far only raw metrics can be obtained via a paged set.
Also updated all the other usage of domain object mappers to the new model.
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateMetricMapper.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateMetricMapper.java index 58642ed..ab66788 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateMetricMapper.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/AggregateMetricMapper.java @@ -25,9 +25,10 @@
package org.rhq.server.metrics;
-import java.sql.ResultSet; -import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List;
+import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row;
/** @@ -45,70 +46,113 @@ public class AggregateMetricMapper implements ResultSetMapper<AggregatedNumericM if (includeMetadata) { resultSetMapper = new ResultSetMapper<AggregatedNumericMetric>() { @Override - public AggregatedNumericMetric map(ResultSet resultSet) throws SQLException { - return null; + public List<AggregatedNumericMetric> mapAll(ResultSet resultSet) { + List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); + while (!resultSet.isExhausted()) { + metrics.add(mapOne(resultSet)); + } + + return metrics; }
@Override - public AggregatedNumericMetric map(Row... row) { - AggregatedNumericMetric metric = new AggregatedNumericMetric(); - metric.setScheduleId(row[0].getInt(0)); - metric.setTimestamp(row[0].getDate(1).getTime()); - metric.setMax(row[0].getDouble(3)); - metric.setMin(row[1].getDouble(3)); - metric.setAvg(row[2].getDouble(3)); - - ColumnMetadata maxMetadata = new ColumnMetadata(row[0].getInt(4), row[0].getLong(5)); - ColumnMetadata minMetadata = new ColumnMetadata(row[1].getInt(4), row[1].getLong(5)); - ColumnMetadata avgMetadata = new ColumnMetadata(row[2].getInt(4), row[2].getLong(5)); - - metric.setAvgColumnMetadata(avgMetadata); - metric.setMaxColumnMetadata(maxMetadata); - metric.setMinColumnMetadata(minMetadata); - - return metric; + public AggregatedNumericMetric mapOne(ResultSet resultSet) { + return map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne()).get(0); + } + + @Override + public List<AggregatedNumericMetric> map(Row... row) { + List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); + + for (int i = 0; i < row.length; i += 3) { + AggregatedNumericMetric metric = new AggregatedNumericMetric(); + metric.setScheduleId(row[i].getInt(0)); + metric.setTimestamp(row[i].getDate(1).getTime()); + metric.setMax(row[i].getDouble(3)); + metric.setMin(row[i + 1].getDouble(3)); + metric.setAvg(row[i + 2].getDouble(3)); + + ColumnMetadata maxMetadata = new ColumnMetadata(row[i].getInt(4), row[i].getLong(5)); + ColumnMetadata minMetadata = new ColumnMetadata(row[i + 1].getInt(4), row[i + 1].getLong(5)); + ColumnMetadata avgMetadata = new ColumnMetadata(row[i + 2].getInt(4), row[i + 2].getLong(5)); + + metric.setAvgColumnMetadata(avgMetadata); + metric.setMaxColumnMetadata(maxMetadata); + metric.setMinColumnMetadata(minMetadata); + + metrics.add(metric); + } + + return metrics; + } + + @Override + public AggregatedNumericMetric map(Row row) { + throw new UnsupportedOperationException( + "Method is not supported. Only triples are accepted for mapping."); } }; } else { resultSetMapper = new ResultSetMapper<AggregatedNumericMetric>() { @Override - public AggregatedNumericMetric map(ResultSet resultSet) throws SQLException { - return null; + public List<AggregatedNumericMetric> mapAll(ResultSet resultSet) { + List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); + while (!resultSet.isExhausted()) { + metrics.add(mapOne(resultSet)); + } + + return metrics; }
@Override - public AggregatedNumericMetric map(Row... row) { - AggregatedNumericMetric metric = new AggregatedNumericMetric(); - metric.setScheduleId(row[0].getInt(0)); - metric.setTimestamp(row[0].getDate(1).getTime()); - metric.setMax(row[0].getDouble(3)); - metric.setMin(row[1].getDouble(3)); - metric.setAvg(row[2].getDouble(3)); - - return metric; + public AggregatedNumericMetric mapOne(ResultSet resultSet) { + return map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne()).get(0); + } + + @Override + public List<AggregatedNumericMetric> map(Row... row) { + List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); + + for (int i = 0; i < row.length; i += 3) { + AggregatedNumericMetric metric = new AggregatedNumericMetric(); + metric.setScheduleId(row[i].getInt(0)); + metric.setTimestamp(row[i].getDate(1).getTime()); + metric.setMax(row[i].getDouble(3)); + metric.setMin(row[i + 1].getDouble(3)); + metric.setAvg(row[i + 2].getDouble(3)); + + metrics.add(metric); + } + + return metrics; + } + + @Override + public AggregatedNumericMetric map(Row row) { + throw new UnsupportedOperationException( + "Method is not supported. Only triples are accepted for mapping."); } }; } }
@Override - public AggregatedNumericMetric map(ResultSet resultSet) throws SQLException { - AggregatedNumericMetric metric = new AggregatedNumericMetric(); - metric.setScheduleId(resultSet.getInt(1)); - metric.setTimestamp(resultSet.getDate(2).getTime()); - metric.setMax(resultSet.getDouble(4)); - - resultSet.next(); - metric.setMin(resultSet.getDouble(4)); - - resultSet.next(); - metric.setAvg(resultSet.getDouble(4)); + public List<AggregatedNumericMetric> mapAll(ResultSet resultSet) { + return resultSetMapper.mapAll(resultSet); + }
- return metric; + @Override + public AggregatedNumericMetric mapOne(ResultSet resultSet) { + return resultSetMapper.mapOne(resultSet); }
@Override - public AggregatedNumericMetric map(Row... rows) { + public List<AggregatedNumericMetric> map(Row... rows) { return resultSetMapper.map(rows); } + + @Override + public AggregatedNumericMetric map(Row row) { + return resultSetMapper.map(row); + } } diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java index 7f940ef..e33db48 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java @@ -221,13 +221,22 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId, new Date(startTime), new Date(endTime)); ResultSet resultSet = session.execute(boundStatement);
- List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); ResultSetMapper<RawNumericMetric> resultSetMapper = new RawNumericMetricMapper(); - for (Row row : resultSet) { - metrics.add(resultSetMapper.map(row)); - } + return resultSetMapper.mapAll(resultSet); + } catch (NoHostAvailableException e) { + throw new CQLException(e); + } + }
- return metrics; + public PagedResultSet<RawNumericMetric> findRawMetricsPaged(int scheduleId, long startTime, long endTime) { + try { + PreparedStatement statement = session.prepare(RAW_METRICS_QUERY); + BoundStatement boundStatement = statement.bind(scheduleId, new Date(startTime), new Date(endTime)); + ResultSet resultSet = session.execute(boundStatement); + + PagedResultSet<RawNumericMetric> pagedResultSet = new PagedResultSet<RawNumericMetric>(resultSet, + new RawNumericMetricMapper()); + return pagedResultSet; } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -243,12 +252,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId); ResultSet resultSet = session.execute(boundStatement);
- List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); ResultSetMapper<RawNumericMetric> resultSetMapper = new RawNumericMetricMapper(); - for (Row row : resultSet) { - metrics.add(resultSetMapper.map(row)); - } - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -266,12 +271,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId, new Date(startTime), new Date(endTime)); ResultSet resultSet = session.execute(boundStatement);
- List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); ResultSetMapper<RawNumericMetric> resultSetMapper = new RawNumericMetricMapper(true); - for (Row row : resultSet) { - metrics.add(resultSetMapper.map(row)); - } - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -288,18 +289,12 @@ public class MetricsDAO { //PreparedStatement statement = session.prepare(RAW_METRICS_SCHEDULE_LIST_QUERY); //BoundStatement boundStatement = statement.bind(scheduleIds, startTime, endTime); //ResultSet resultSet = session.execute(boundStatement); - String cql = "SELECT schedule_id, time, value FROM " + MetricsTable.RAW + " WHERE schedule_id IN (" + listToString(scheduleIds) + ") AND time >= " + startTime + " AND time <= " + endTime; ResultSet resultSet = session.execute(cql);
- List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); ResultSetMapper<RawNumericMetric> resultSetMapper = new RawNumericMetricMapper(false); - for (Row row : resultSet) { - metrics.add(resultSetMapper.map(row)); - } - - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -316,13 +311,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId); ResultSet resultSet = session.execute(boundStatement);
- List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); ResultSetMapper<AggregatedNumericMetric> resultSetMapper = new AggregateMetricMapper(); - while (!resultSet.isExhausted()) { - metrics.add(resultSetMapper.map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne())); - } - - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -340,13 +330,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId, new Date(startTime), new Date(endTime)); ResultSet resultSet = session.execute(boundStatement);
- List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); ResultSetMapper<AggregatedNumericMetric> resultSetMapper = new AggregateMetricMapper(); - while (!resultSet.isExhausted()) { - metrics.add(resultSetMapper.map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne())); - } - - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -380,13 +365,8 @@ public class MetricsDAO { "WHERE schedule_id IN (" + listToString(scheduleIds) + ") AND time >= " + startTime + " AND time < " + endTime; ResultSet resultSet = session.execute(cql);
- List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); ResultSetMapper<AggregatedNumericMetric> resultSetMapper = new AggregateMetricMapper(); - while (!resultSet.isExhausted()) { - metrics.add(resultSetMapper.map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne())); - } - - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -404,13 +384,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(scheduleId, new Date(startTime), new Date(endTime)); ResultSet resultSet = session.execute(boundStatement);
- List<AggregatedNumericMetric> metrics = new ArrayList<AggregatedNumericMetric>(); ResultSetMapper<AggregatedNumericMetric> resultSetMapper = new AggregateMetricMapper(true); - while (!resultSet.isExhausted()) { - metrics.add(resultSetMapper.map(resultSet.fetchOne(), resultSet.fetchOne(), resultSet.fetchOne())); - } - - return metrics; + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } @@ -422,14 +397,8 @@ public class MetricsDAO { BoundStatement boundStatement = statement.bind(table.toString()); ResultSet resultSet = session.execute(boundStatement);
- List<MetricsIndexEntry> indexEntries = new ArrayList<MetricsIndexEntry>(); ResultSetMapper<MetricsIndexEntry> resultSetMapper = new MetricsIndexResultSetMapper(table); - for (Row row : resultSet) { - indexEntries.add(resultSetMapper.map(row)); - } - - return indexEntries; - + return resultSetMapper.mapAll(resultSet); } catch (NoHostAvailableException e) { throw new CQLException(e); } diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexResultSetMapper.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexResultSetMapper.java index 3a0db3e..5c9a4c3 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexResultSetMapper.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsIndexResultSetMapper.java @@ -25,9 +25,10 @@
package org.rhq.server.metrics;
-import java.sql.ResultSet; -import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List;
+import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row;
@@ -43,12 +44,32 @@ public class MetricsIndexResultSetMapper implements ResultSetMapper<MetricsIndex }
@Override - public MetricsIndexEntry map(ResultSet resultSet) throws SQLException { - return new MetricsIndexEntry(bucket, resultSet.getDate(1), resultSet.getInt(2)); + public List<MetricsIndexEntry> mapAll(ResultSet resultSet) { + List<MetricsIndexEntry> result = new ArrayList<MetricsIndexEntry>(); + for (Row singleRow : resultSet) { + result.add(map(singleRow)); + } + + return result; + } + + @Override + public MetricsIndexEntry mapOne(ResultSet resultSet) { + return map(resultSet.fetchOne()); + } + + @Override + public List<MetricsIndexEntry> map(Row... row) { + List<MetricsIndexEntry> result = new ArrayList<MetricsIndexEntry>(); + for (Row singleRow : row) { + result.add(new MetricsIndexEntry(bucket, singleRow.getDate(0), singleRow.getInt(1))); + } + + return result; }
@Override - public MetricsIndexEntry map(Row... row) { - return new MetricsIndexEntry(bucket, row[0].getDate(0), row[0].getInt(1)); + public MetricsIndexEntry map(Row row) { + return new MetricsIndexEntry(bucket, row.getDate(0), row.getInt(1)); } } diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/PagedResultSet.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/PagedResultSet.java new file mode 100644 index 0000000..be0145a --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/PagedResultSet.java @@ -0,0 +1,118 @@ +/* + * RHQ Management Platform + * Copyright 2011, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation version 2 of the License. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +package org.rhq.server.metrics; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +import com.datastax.driver.core.ResultSet; + +/** + * + * @author Stefan Negrea + * + */ +public class PagedResultSet<T> implements Iterable<List<T>> { + + private static final int DEFAULT_PAGE_SIZE = 200; + + private final ResultSetMapper<T> mapper; + private final ResultSet resultSet; + + private int pageSize = DEFAULT_PAGE_SIZE; + + /** + * @param resultSet result set to map + * @param mapper mapper + * @param pageSize page size + */ + public PagedResultSet(ResultSet resultSet, ResultSetMapper<T> mapper, int pageSize) { + this(resultSet, mapper); + this.pageSize = pageSize; + } + + /** + * @param resultSet result set to map + * @param mapper mapper + */ + public PagedResultSet(ResultSet resultSet, ResultSetMapper<T> mapper) { + this.resultSet = resultSet; + this.mapper = mapper; + } + + /** + * @return the page size + */ + public int getPageSize() { + return pageSize; + } + + /** + * @param pageSize the page size to set + */ + public void setPageSize(int pageSize) { + this.pageSize = pageSize; + } + + public List<T> getNextPage() { + List<T> result = new ArrayList<T>(); + for (int i = 0; i < this.getPageSize(); i++) { + if (!resultSet.isExhausted()) { + result.add((T) mapper.mapOne(resultSet)); + } else { + break; + } + } + + return null; + } + + /** + * @return true if the result is exhausted, false otherwise + */ + public boolean isExhausted() { + return resultSet.isExhausted(); + } + + /* (non-Javadoc) + * @see java.lang.Iterable#iterator() + */ + @Override + public Iterator<List<T>> iterator() { + + return new Iterator<List<T>>() { + + public boolean hasNext() { + return !isExhausted(); + } + + public List<T> next() { + return getNextPage(); + } + + public void remove() { + throw new UnsupportedOperationException(); + } + }; + } + +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Query.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Query.java index 938feea..17ec498 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Query.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/Query.java @@ -54,8 +54,8 @@ public class Query<T> { try { resultSet = statement.executeQuery(); while (resultSet.next()) { - T row = resultSetMapper.map(resultSet); - callback.invoke(row); + //T row = resultSetMapper.map(resultSet); + //callback.invoke(row); } } catch(SQLException e) { throw new CQLException(e); diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawNumericMetricMapper.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawNumericMetricMapper.java index f697543..329a20e 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawNumericMetricMapper.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawNumericMetricMapper.java @@ -25,9 +25,10 @@
package org.rhq.server.metrics;
-import java.sql.ResultSet; -import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List;
+import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row;
/** @@ -43,60 +44,101 @@ public class RawNumericMetricMapper implements ResultSetMapper<RawNumericMetric>
public RawNumericMetricMapper(boolean metaDataIncluded) { if (metaDataIncluded) { -// mapper = new ResultSetMapper<RawNumericMetric>() { -// @Override -// public RawNumericMetric map(ResultSet resultSet) throws SQLException { -// RawNumericMetric rawMetric = new RawNumericMetric(resultSet.getInt(1), -// resultSet.getDate(2).getTime(), resultSet.getDouble(3)); -// ColumnMetadata metadata = new ColumnMetadata(resultSet.getInt(4), resultSet.getLong(5)); -// rawMetric.setColumnMetadata(metadata); -// return rawMetric; -// } -// }; mapper = new ResultSetMapper<RawNumericMetric>() { @Override - public RawNumericMetric map(ResultSet resultSet) throws SQLException { - return null; + public List<RawNumericMetric> mapAll(ResultSet resultSet) { + List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); + for (Row row : resultSet) { + metrics.add(map(row)); + } + + return metrics; + } + + @Override + public RawNumericMetric mapOne(ResultSet resultSet) { + return mapper.map(resultSet.fetchOne()); }
@Override - public RawNumericMetric map(Row... row) { - RawNumericMetric metric = new RawNumericMetric(row[0].getInt(0), row[0].getDate(1).getTime(), - row[0].getDouble(2)); - ColumnMetadata metadata = new ColumnMetadata(row[0].getInt(3), row[0].getLong(4)); + public List<RawNumericMetric> map(Row... row) { + List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); + + for (Row singleRow : row) { + RawNumericMetric metric = new RawNumericMetric(singleRow.getInt(0), singleRow.getDate(1) + .getTime(), singleRow.getDouble(2)); + ColumnMetadata metadata = new ColumnMetadata(singleRow.getInt(3), singleRow.getLong(4)); + metric.setColumnMetadata(metadata); + + metrics.add(metric); + } + + return metrics; + } + + @Override + public RawNumericMetric map(Row row) { + RawNumericMetric metric = new RawNumericMetric(row.getInt(0), row.getDate(1).getTime(), + row.getDouble(2)); + ColumnMetadata metadata = new ColumnMetadata(row.getInt(3), row.getLong(4)); metric.setColumnMetadata(metadata); + return metric; } }; } else { -// mapper = new ResultSetMapper<RawNumericMetric>() { -// @Override -// public RawNumericMetric map(ResultSet resultSet) throws SQLException { -// return new RawNumericMetric(resultSet.getInt(1), resultSet.getDate(2).getTime(), -// resultSet.getDouble(3)); -// } -// }; mapper = new ResultSetMapper<RawNumericMetric>() { @Override - public RawNumericMetric map(ResultSet resultSet) throws SQLException { - return null; + public List<RawNumericMetric> mapAll(ResultSet resultSet) { + List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); + for (Row row : resultSet) { + metrics.add(map(row)); + } + + return metrics; + } + + @Override + public RawNumericMetric mapOne(ResultSet resultSet) { + return map(resultSet.fetchOne()); }
@Override - public RawNumericMetric map(Row... row) { - return new RawNumericMetric(row[0].getInt(0), row[0].getDate(1).getTime(), row[0].getDouble(2)); + public List<RawNumericMetric> map(Row... row) { + List<RawNumericMetric> metrics = new ArrayList<RawNumericMetric>(); + for (Row singleRow : row) { + metrics.add(new RawNumericMetric(singleRow.getInt(0), singleRow.getDate(1).getTime(), singleRow + .getDouble(2))); + } + + return metrics; + } + + @Override + public RawNumericMetric map(Row row) { + return new RawNumericMetric(row.getInt(0), row.getDate(1).getTime(), row.getDouble(2)); } }; } }
@Override - public RawNumericMetric map(ResultSet resultSet) throws SQLException { - return mapper.map(resultSet); + public List<RawNumericMetric> mapAll(ResultSet resultSet) { + return mapper.mapAll(resultSet); + } + + @Override + public RawNumericMetric mapOne(ResultSet resultSet) { + return mapper.mapOne(resultSet); + } + + @Override + public List<RawNumericMetric> map(Row... row) { + return mapper.map(row); }
@Override - public RawNumericMetric map(Row... row) { + public RawNumericMetric map(Row row) { return mapper.map(row); } } diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ResultSetMapper.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ResultSetMapper.java index f62af8a..c1ea60e 100644 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ResultSetMapper.java +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ResultSetMapper.java @@ -25,9 +25,9 @@
package org.rhq.server.metrics;
-import java.sql.ResultSet; -import java.sql.SQLException; +import java.util.List;
+import com.datastax.driver.core.ResultSet; import com.datastax.driver.core.Row;
/** @@ -35,8 +35,37 @@ import com.datastax.driver.core.Row; */ public interface ResultSetMapper<T> {
- T map(ResultSet resultSet) throws SQLException; + /** + * Maps the entire result set to a list. + * + * @param resultSet result set to map + * @return a list of mapped rows + */ + List<T> mapAll(ResultSet resultSet);
- T map(Row... row); + /** + * Return only one mapped domain object. One mapped object could + * require multiple rows from the result set. + * + * @param resultSet result set to map + * @return + */ + T mapOne(ResultSet resultSet);
+ /** + * Map all the rows passed to domain objects. More than one row could + * be required for mapping a single object. + * + * @param row + * @return + */ + List<T> map(Row... row); + + /** + * Map a sigle row to a domain object. + * + * @param row + * @return + */ + T map(Row row); }
rhq-commits@lists.fedorahosted.org