DatastreamRestClient { private List<Datastream> getAllDatastreams(GetAllRequest<Datastream> request) { Instant startTime = Instant.now(); List<Datastream> datastreams = PollUtils.poll(() -> { ResponseFuture<CollectionResponse<Datastream>> datastreamResponseFuture = _restClient.sendRequest(request); try { return datastreamResponseFuture.getResponse().getEntity().getElements(); } catch (RemoteInvocationException e) { if (ExceptionUtils.getRootCause(e) instanceof TimeoutException) { LOG.warn("Timeout: getAllDatastreams. May retry...", e); return null; } String errorMessage = "Get All Datastreams failed with error."; ErrorLogger.logAndThrowDatastreamRuntimeException(LOG, errorMessage, e); return null; } }, Objects::nonNull, getRetryPeriodMs(), getRetryTimeoutMs()).orElseThrow(RetriesExhaustedException::new); LOG.info("getAllDatastreams took {} ms", Duration.between(startTime, Instant.now()).toMillis()); return datastreams; } @Deprecated  DatastreamRestClient(RestClient restClient);  DatastreamRestClient(RestClient restClient, Properties config); Datastream getDatastream(String datastreamName); Datastream waitTillDatastreamIsInitialized(String datastreamName, long timeoutMs); void waitTillDatastreamIsDeleted(String datastreamName, long timeoutMs); List<Datastream> getAllDatastreams(); List<Datastream> getAllDatastreams(int start, int count); void createDatastream(Datastream datastream); void updateDatastream(Datastream datastream); void updateDatastream(List<Datastream> datastreams); void deleteDatastream(String datastreamName); boolean datastreamExists(String datastreamName); void pause(String datastreamName); void pause(String datastreamName, boolean force); void stop(String datastreamName); void resume(String datastreamName); void resume(String datastreamName, boolean force); void movePartitions(String datastreamName, String partitions, String targetHost); List<Datastream> findGroup(String datastreamName); void pauseSourcePartitions(String datastreamName, StringMap sourcePartitions); void resumeSourcePartitions(String datastreamName, StringMap sourcePartitions); static final String CONFIG_RETRY_PERIOD_MS; static final String CONFIG_RETRY_TIMEOUT_MS; }
@Test public void testGetAllDatastreams() { List<Datastream> datastreams = IntStream.range(100, 110).mapToObj(TestDatastreamRestClient::generateDatastream).collect(Collectors.toList()); LOG.info("Datastreams : {}", datastreams); DatastreamRestClient restClient = createRestClient(); int initialSize = restClient.getAllDatastreams().size(); int createdCount = datastreams.size(); for (Datastream datastream : datastreams) { restClient.createDatastream(datastream); } Optional<List<Datastream>> result = PollUtils.poll(restClient::getAllDatastreams, streams -> streams.size() - initialSize == createdCount, 100, 1000); Assert.assertTrue(result.isPresent()); List<Datastream> createdDatastreams = result.get(); LOG.info("Created Datastreams : {}", createdDatastreams); clearDatastreamDestination(datastreams); clearDatastreamDestination(createdDatastreams); clearDynamicMetadata(createdDatastreams); Assert.assertTrue(new HashSet<>(createdDatastreams).containsAll(datastreams), "Original datastreams " + datastreams + " not present in last getAll " + createdDatastreams); int skip = 2; int count = 5; List<Datastream> paginatedCreatedDatastreams = restClient.getAllDatastreams(2, 5); LOG.info("Paginated Datastreams : {}", paginatedCreatedDatastreams); Assert.assertEquals(paginatedCreatedDatastreams.size(), count); clearDatastreamDestination(paginatedCreatedDatastreams); clearDynamicMetadata(paginatedCreatedDatastreams); Assert.assertEquals(createdDatastreams.stream().skip(skip).limit(count).collect(Collectors.toList()), paginatedCreatedDatastreams); }