FileMetadataRegistryOnPostgres implements FileMetadataRegistry { @Override public Observable<FileMetadata> iterate() { return postgreSQLClient.getConnectionObservable() .flatMap(con -> { return con.queryObservable("SELECT id, name, mime_type, content_length, generated from file_metadata ORDER BY generated DESC") .doAfterTerminate(con::close); }).flatMap(rs -> { List<FileMetadata> iterable = rs.getResults().stream().map(json -> { @FileId UUID fileId = UUID.fromString(json.getString(0)); MimeType mimeType = MimeType.valueOf(json.getString(2)); Instant generated = Instant.parse(json.getString(4) + "Z"); return new FileMetadata(fileId, json.getString(1), mimeType, json.getLong(3), generated); }).collect(Collectors.toList()); return Observable.from(iterable); }); }  @Override Observable<FileMetadata> iterate(); @Override Single<FileMetadata> store(FileMetadata metadata); @Override Single<FileMetadata> update(FileMetadata metadata); @Override Single<FileMetadata> load(UUID fileId); @Override Single<UUID> delete(UUID fileId);  }
@Test public void testIterate(TestContext context) { UUID fileId = UUID.randomUUID(); Async async = context.async(2); FileMetadata metadata = new FileMetadata(fileId, "name", MimeType.valueOf("text/plain"), 6, Instant.now()); registry.store(metadata).toObservable().flatMap(v -> { return registry.iterate(); }).filter(iterated -> { return iterated.equals(metadata); }).count().subscribe(count -> { context.assertEquals(1, count); async.complete(); }, context::fail, async::complete); }