SampleContinuousQueryObserverService extends StatefulService { public SampleContinuousQueryObserverService() { super(QueryObserverState.class); toggleOption(ServiceOption.INSTRUMENTATION, true); toggleOption(ServiceOption.PERSISTENCE, true); } SampleContinuousQueryObserverService(); @Override void handleStart(Operation startPost); @Override void handlePatch(Operation patch); static final String FACTORY_LINK; }
@Test public void testSampleContinuousQueryObserverService() throws Throwable { QueryObserverState queryObserverInitialState = new QueryObserverState(); queryObserverInitialState.notificationsCounter = 0; Operation post = Operation.createPost(UriUtils.buildUri(this.host.getUri(), SampleContinuousQueryObserverService.FACTORY_LINK)) .setBody(queryObserverInitialState) .setReferer(this.host.getUri()); queryObserverInitialState = this.sender.sendAndWait(post, QueryObserverState.class); final String queryObserverSelfLink = queryObserverInitialState.documentSelfLink; waitFor(this.timeout, () -> { ServiceStats indexStats = this.host.getServiceState(null, ServiceStats.class, UriUtils.buildStatsUri(this.host.getDocumentIndexServiceUri())); ServiceStat activeQueryStat = indexStats.entries.get( LuceneDocumentIndexService.STAT_NAME_ACTIVE_QUERY_FILTERS + ServiceStats.STAT_NAME_SUFFIX_PER_DAY); if (activeQueryStat == null || activeQueryStat.latestValue < 1.0) { return false; } return true; }); List<String> samplePreviousEchoServicesLinks = new ArrayList<>(); for (int i = 0; i < this.serviceCount; i++) { EchoServiceState state = new EchoServiceState(); state.message = "hello"; post = Operation.createPost(UriUtils.buildUri(this.host.getUri(), SamplePreviousEchoService.FACTORY_LINK)) .setBody(state) .setReferer(this.host.getUri()); state = this.sender.sendAndWait(post, EchoServiceState.class); samplePreviousEchoServicesLinks.add(state.documentSelfLink); } waitFor(this.timeout, () -> { Operation getQueryObserverState = Operation.createGet( UriUtils.buildUri(this.host.getUri(), queryObserverSelfLink)) .setReferer(this.host.getUri()); QueryObserverState updatedQueryObserverState = this.sender.sendAndWait(getQueryObserverState, QueryObserverState.class); this.host.log("notification count: %d", updatedQueryObserverState.notificationsCounter); return (this.serviceCount <= updatedQueryObserverState.notificationsCounter); }); final EchoServiceState updatedState = new EchoServiceState(); updatedState.message = "hello world"; TestContext ctx1 = new TestContext(samplePreviousEchoServicesLinks.size(), this.timeout); samplePreviousEchoServicesLinks.forEach(selfLink -> { Operation putOp = Operation.createPut(UriUtils.buildUri(this.host.getUri(), selfLink)) .setBody(updatedState) .setReferer(this.host.getUri()) .setCompletion(ctx1.getCompletion()); this.sender.sendRequest(putOp); }); ctx1.await(); waitFor(this.timeout, () -> { Operation getQueryObserverState = Operation.createGet( UriUtils.buildUri(this.host.getUri(), queryObserverSelfLink)) .setReferer(this.host.getUri()); QueryObserverState updatedQueryObserverState = this.sender.sendAndWait(getQueryObserverState, QueryObserverState.class); this.host.log("notification count: %d", updatedQueryObserverState.notificationsCounter); return (this.serviceCount * 2 <= updatedQueryObserverState.notificationsCounter); }); TestContext ctx = new TestContext(samplePreviousEchoServicesLinks.size(), this.timeout); samplePreviousEchoServicesLinks.forEach(selfLink -> { Operation delete = Operation .createDelete(UriUtils.buildUri(this.host.getUri(), selfLink)) .setReferer(this.host.getUri()).setCompletion(ctx.getCompletion()); this.sender.sendRequest(delete); }); ctx.await(); waitFor(this.timeout, () -> { Operation getQueryObserverState = Operation.createGet( UriUtils.buildUri(this.host.getUri(), queryObserverSelfLink)) .setReferer(this.host.getUri()); QueryObserverState updatedQueryObserverState = this.sender.sendAndWait(getQueryObserverState, QueryObserverState.class); this.host.log("notification count: %d", updatedQueryObserverState.notificationsCounter); return (this.serviceCount * 3 <= updatedQueryObserverState.notificationsCounter); }); }