@Test public void testFindEntities() throws Exception { final List<Criteria.Field> fr = new ArrayList<>(); fr.add(new CriteriaImpl.FieldImpl("ruleId", "1")); fr.add(new CriteriaImpl.FieldImpl("status", "NEW")); final Map<byte[], byte[]> tsMap = new TreeMap<>(new Bytes.ByteArrayComparator()); tsMap.put("1444042473518".getBytes(), "1".getBytes()); final Map<byte[], byte[]> niMap = new TreeMap<>(new Bytes.ByteArrayComparator()); niMap.put("nid".getBytes(), "1".getBytes()); final List<Result> results = Arrays.asList(mockResult); new Expectations() { { mockCriteria.clazz(); times = 1; result = Notification.class; mockCriteria.fieldRestrictions(); times = 1; result = fr; mockHTable.getScanner(withAny(new Scan())); times = 1; result = mockResultScanner; mockResultScanner.iterator(); result = results.iterator(); mockResult.getFamilyMap("ni".getBytes()); times = 1; result = niMap; mockResult.getValue("s".getBytes(), "qs".getBytes()); times = 1; result = "NEW".getBytes(); mockResult.getFamilyMap("ts".getBytes()); times = 1; result = tsMap; } }; List<Notification> notifications = notificationStore.findEntities(mockCriteria); System.out.println(notifications); assertEquals("nid", notifications.get(0).getId()); assertEquals(Notification.Status.NEW, notifications.get(0).getStatus()); assertEquals(1444042473518L, notifications.get(0).getTs()); new Verifications() { { Scan scan; mockHTable.getScanner(scan = withCapture()); times = 1; System.out.println("Scan = " + scan); } }; }
@Override public <T> List<T> findEntities(Criteria<T> criteria) { List<T> entities = new ArrayList<>(); LOG.debug("Finding entities from HBaseNotificationStore, Criteria {}", criteria); try { HBaseScanConfig<T> scanConfig = hBaseScanConfigBuilder.getScanConfig(criteria); LOG.debug("HBaseScanConfig for scan {}", scanConfig); if (scanConfig != null) { byte[] startRow = scanConfig.getStartRow(); byte[] stopRow = scanConfig.getStopRow(); Scan scan; if(criteria.isDescending()) { scan = new Scan(stopRow, startRow); scan.setReversed(true); } else { scan = new Scan(startRow, stopRow); } scan.setFilter(scanConfig.filterList()); ResultScanner scanner = tables.get(scanConfig.getMapper().getTableName()).get().getScanner(scan); for (Result result : scanner) { entities.add(scanConfig.getMapper().entity(result)); } } } catch (IOException ex) { throw new NotificationStoreException("Error during scan", ex); } return entities; }
HBaseNotificationStore implements NotificationStore { @Override public <T> List<T> findEntities(Criteria<T> criteria) { List<T> entities = new ArrayList<>(); LOG.debug("Finding entities from HBaseNotificationStore, Criteria {}", criteria); try { HBaseScanConfig<T> scanConfig = hBaseScanConfigBuilder.getScanConfig(criteria); LOG.debug("HBaseScanConfig for scan {}", scanConfig); if (scanConfig != null) { byte[] startRow = scanConfig.getStartRow(); byte[] stopRow = scanConfig.getStopRow(); Scan scan; if(criteria.isDescending()) { scan = new Scan(stopRow, startRow); scan.setReversed(true); } else { scan = new Scan(startRow, stopRow); } scan.setFilter(scanConfig.filterList()); ResultScanner scanner = tables.get(scanConfig.getMapper().getTableName()).get().getScanner(scan); for (Result result : scanner) { entities.add(scanConfig.getMapper().entity(result)); } } } catch (IOException ex) { throw new NotificationStoreException("Error during scan", ex); } return entities; } }
HBaseNotificationStore implements NotificationStore { @Override public <T> List<T> findEntities(Criteria<T> criteria) { List<T> entities = new ArrayList<>(); LOG.debug("Finding entities from HBaseNotificationStore, Criteria {}", criteria); try { HBaseScanConfig<T> scanConfig = hBaseScanConfigBuilder.getScanConfig(criteria); LOG.debug("HBaseScanConfig for scan {}", scanConfig); if (scanConfig != null) { byte[] startRow = scanConfig.getStartRow(); byte[] stopRow = scanConfig.getStopRow(); Scan scan; if(criteria.isDescending()) { scan = new Scan(stopRow, startRow); scan.setReversed(true); } else { scan = new Scan(startRow, stopRow); } scan.setFilter(scanConfig.filterList()); ResultScanner scanner = tables.get(scanConfig.getMapper().getTableName()).get().getScanner(scan); for (Result result : scanner) { entities.add(scanConfig.getMapper().entity(result)); } } } catch (IOException ex) { throw new NotificationStoreException("Error during scan", ex); } return entities; }  }
HBaseNotificationStore implements NotificationStore { @Override public <T> List<T> findEntities(Criteria<T> criteria) { List<T> entities = new ArrayList<>(); LOG.debug("Finding entities from HBaseNotificationStore, Criteria {}", criteria); try { HBaseScanConfig<T> scanConfig = hBaseScanConfigBuilder.getScanConfig(criteria); LOG.debug("HBaseScanConfig for scan {}", scanConfig); if (scanConfig != null) { byte[] startRow = scanConfig.getStartRow(); byte[] stopRow = scanConfig.getStopRow(); Scan scan; if(criteria.isDescending()) { scan = new Scan(stopRow, startRow); scan.setReversed(true); } else { scan = new Scan(startRow, stopRow); } scan.setFilter(scanConfig.filterList()); ResultScanner scanner = tables.get(scanConfig.getMapper().getTableName()).get().getScanner(scan); for (Result result : scanner) { entities.add(scanConfig.getMapper().entity(result)); } } } catch (IOException ex) { throw new NotificationStoreException("Error during scan", ex); } return entities; }  @Override void init(Map<String, Object> config); @Override void store(Notification notification); @Override Notification getNotification(String notificationId); @Override List<Notification> getNotifications(List<String> notificationIds); @Override StreamlineEvent getEvent(String eventId); @Override List<StreamlineEvent> getEvents(List<String> eventIds); @Override List<T> findEntities(Criteria<T> criteria); @Override void close(); @Override Notification updateNotificationStatus(String notificationId, Notification.Status status); }
HBaseNotificationStore implements NotificationStore { @Override public <T> List<T> findEntities(Criteria<T> criteria) { List<T> entities = new ArrayList<>(); LOG.debug("Finding entities from HBaseNotificationStore, Criteria {}", criteria); try { HBaseScanConfig<T> scanConfig = hBaseScanConfigBuilder.getScanConfig(criteria); LOG.debug("HBaseScanConfig for scan {}", scanConfig); if (scanConfig != null) { byte[] startRow = scanConfig.getStartRow(); byte[] stopRow = scanConfig.getStopRow(); Scan scan; if(criteria.isDescending()) { scan = new Scan(stopRow, startRow); scan.setReversed(true); } else { scan = new Scan(startRow, stopRow); } scan.setFilter(scanConfig.filterList()); ResultScanner scanner = tables.get(scanConfig.getMapper().getTableName()).get().getScanner(scan); for (Result result : scanner) { entities.add(scanConfig.getMapper().entity(result)); } } } catch (IOException ex) { throw new NotificationStoreException("Error during scan", ex); } return entities; }  @Override void init(Map<String, Object> config); @Override void store(Notification notification); @Override Notification getNotification(String notificationId); @Override List<Notification> getNotifications(List<String> notificationIds); @Override StreamlineEvent getEvent(String eventId); @Override List<StreamlineEvent> getEvents(List<String> eventIds); @Override List<T> findEntities(Criteria<T> criteria); @Override void close(); @Override Notification updateNotificationStatus(String notificationId, Notification.Status status);  }