@Test public void removeAcknowledged() throws ChannelNotFoundException { final String uaid = UUIDUtil.newUAID(); final Channel channel1 = newChannel(uaid, UUID.randomUUID().toString(), 10); final Channel channel2 = newChannel(uaid, UUID.randomUUID().toString(), 22); datastore.saveChannel(channel1); datastore.saveChannel(channel2); datastore.saveUnacknowledged(channel1.getChannelId(), channel1.getVersion()); datastore.saveUnacknowledged(channel2.getChannelId(), channel2.getVersion()); final Set<Ack> unacks = datastore.removeAcknowledged(uaid, acks(ack(channel1))); assertThat(unacks, hasItem(ack(channel2))); assertThat(unacks.size(), is(1)); assertThat(datastore.removeAcknowledged(uaid, unacks).size(), is(0)); }
@Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked) { final ViewResult viewResult = db.queryView(query(Views.UNACKS.viewName(), uaid)); final List<Row> rows = viewResult.getRows(); final Collection<BulkDeleteDocument> removals = new LinkedHashSet<BulkDeleteDocument>(); for (Iterator<Row> iter = rows.iterator(); iter.hasNext(); ) { final Row row = iter.next(); final JsonNode json = row.getValueAsNode(); final JsonNode doc = json.get(DOC_FIELD); final String channelId = doc.get(CHID_FIELD).asText(); for (Ack ack : acked) { if (ack.getChannelId().equals(channelId)) { removals.add(BulkDeleteDocument.of(doc)); iter.remove(); } } } db.executeBulk(removals); return rowsToAcks(rows); }
CouchDBDataStore implements DataStore { @Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked) { final ViewResult viewResult = db.queryView(query(Views.UNACKS.viewName(), uaid)); final List<Row> rows = viewResult.getRows(); final Collection<BulkDeleteDocument> removals = new LinkedHashSet<BulkDeleteDocument>(); for (Iterator<Row> iter = rows.iterator(); iter.hasNext(); ) { final Row row = iter.next(); final JsonNode json = row.getValueAsNode(); final JsonNode doc = json.get(DOC_FIELD); final String channelId = doc.get(CHID_FIELD).asText(); for (Ack ack : acked) { if (ack.getChannelId().equals(channelId)) { removals.add(BulkDeleteDocument.of(doc)); iter.remove(); } } } db.executeBulk(removals); return rowsToAcks(rows); } }
CouchDBDataStore implements DataStore { @Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked) { final ViewResult viewResult = db.queryView(query(Views.UNACKS.viewName(), uaid)); final List<Row> rows = viewResult.getRows(); final Collection<BulkDeleteDocument> removals = new LinkedHashSet<BulkDeleteDocument>(); for (Iterator<Row> iter = rows.iterator(); iter.hasNext(); ) { final Row row = iter.next(); final JsonNode json = row.getValueAsNode(); final JsonNode doc = json.get(DOC_FIELD); final String channelId = doc.get(CHID_FIELD).asText(); for (Ack ack : acked) { if (ack.getChannelId().equals(channelId)) { removals.add(BulkDeleteDocument.of(doc)); iter.remove(); } } } db.executeBulk(removals); return rowsToAcks(rows); } CouchDBDataStore(final String url, final String dbName); }
CouchDBDataStore implements DataStore { @Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked) { final ViewResult viewResult = db.queryView(query(Views.UNACKS.viewName(), uaid)); final List<Row> rows = viewResult.getRows(); final Collection<BulkDeleteDocument> removals = new LinkedHashSet<BulkDeleteDocument>(); for (Iterator<Row> iter = rows.iterator(); iter.hasNext(); ) { final Row row = iter.next(); final JsonNode json = row.getValueAsNode(); final JsonNode doc = json.get(DOC_FIELD); final String channelId = doc.get(CHID_FIELD).asText(); for (Ack ack : acked) { if (ack.getChannelId().equals(channelId)) { removals.add(BulkDeleteDocument.of(doc)); iter.remove(); } } } db.executeBulk(removals); return rowsToAcks(rows); } CouchDBDataStore(final String url, final String dbName); @Override void savePrivateKeySalt(final byte[] salt); @Override byte[] getPrivateKeySalt(); @Override boolean saveChannel(final Channel channel); @Override Channel getChannel(final String channelId); @Override void removeChannels(final String uaid); @Override void removeChannels(final Set<String> channelIds); @Override Set<String> getChannelIds(final String uaid); @Override String updateVersion(final String endpointToken, final long version); @Override String saveUnacknowledged(final String channelId, final long version); @Override Set<Ack> getUnacknowledged(final String uaid); @Override Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked); }
CouchDBDataStore implements DataStore { @Override public Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked) { final ViewResult viewResult = db.queryView(query(Views.UNACKS.viewName(), uaid)); final List<Row> rows = viewResult.getRows(); final Collection<BulkDeleteDocument> removals = new LinkedHashSet<BulkDeleteDocument>(); for (Iterator<Row> iter = rows.iterator(); iter.hasNext(); ) { final Row row = iter.next(); final JsonNode json = row.getValueAsNode(); final JsonNode doc = json.get(DOC_FIELD); final String channelId = doc.get(CHID_FIELD).asText(); for (Ack ack : acked) { if (ack.getChannelId().equals(channelId)) { removals.add(BulkDeleteDocument.of(doc)); iter.remove(); } } } db.executeBulk(removals); return rowsToAcks(rows); } CouchDBDataStore(final String url, final String dbName); @Override void savePrivateKeySalt(final byte[] salt); @Override byte[] getPrivateKeySalt(); @Override boolean saveChannel(final Channel channel); @Override Channel getChannel(final String channelId); @Override void removeChannels(final String uaid); @Override void removeChannels(final Set<String> channelIds); @Override Set<String> getChannelIds(final String uaid); @Override String updateVersion(final String endpointToken, final long version); @Override String saveUnacknowledged(final String channelId, final long version); @Override Set<Ack> getUnacknowledged(final String uaid); @Override Set<Ack> removeAcknowledged(final String uaid, final Set<Ack> acked);  }