@Test public void download_ipv4() { BgpRisDump<BgpRisEntry> dump = bgpRisDownloader.fetch( BgpRisDump.of("http: Stream::of ); assertThat(dump.getEntries()).hasValueSatisfying(entries -> assertThat(entries.size()).isGreaterThan(800_000)); }
public <T> BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper) { log.info("attempting to download new BGP RIS preview dump from {}", dump.url); long before = System.currentTimeMillis(); String statusDescription = "200"; final Supplier<Request> requestSupplier = () -> { final Request request = httpClient.newRequest(dump.url); if (dump.lastModified != null) { log.debug("Adding 'If-Modified-Since' equals to {}", formatAsRFC2616(dump.lastModified)); request.header("If-Modified-Since", formatAsRFC2616(dump.lastModified)); } return request; }; final BiFunction<InputStream, Long, BgpRisDump> streamReader = (stream, lastModified) -> { try { Stream<BgpRisEntry> entries = parse(new GZIPInputStream(stream)); ImmutableList.Builder<T> builder = ImmutableList.builder(); entries.flatMap(mapper).forEach(builder::add); return BgpRisDump.of(dump.url, new DateTime(lastModified), Optional.of(builder.build())); } catch (Exception e) { log.error("Error downloading RIS dump: " + dump.url); return dump; } }; try { return HttpStreaming.readStream(requestSupplier, streamReader); } catch (HttpStreaming.NotModifiedException n) { statusDescription = "302"; return dump; } catch (Exception e) { statusDescription = HttpClientMetricsService.unwrapExceptionString(e); throw e; } finally { httpMetrics.update(dump.url, statusDescription, System.currentTimeMillis() - before); } }
BgpRisDownloader { public <T> BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper) { log.info("attempting to download new BGP RIS preview dump from {}", dump.url); long before = System.currentTimeMillis(); String statusDescription = "200"; final Supplier<Request> requestSupplier = () -> { final Request request = httpClient.newRequest(dump.url); if (dump.lastModified != null) { log.debug("Adding 'If-Modified-Since' equals to {}", formatAsRFC2616(dump.lastModified)); request.header("If-Modified-Since", formatAsRFC2616(dump.lastModified)); } return request; }; final BiFunction<InputStream, Long, BgpRisDump> streamReader = (stream, lastModified) -> { try { Stream<BgpRisEntry> entries = parse(new GZIPInputStream(stream)); ImmutableList.Builder<T> builder = ImmutableList.builder(); entries.flatMap(mapper).forEach(builder::add); return BgpRisDump.of(dump.url, new DateTime(lastModified), Optional.of(builder.build())); } catch (Exception e) { log.error("Error downloading RIS dump: " + dump.url); return dump; } }; try { return HttpStreaming.readStream(requestSupplier, streamReader); } catch (HttpStreaming.NotModifiedException n) { statusDescription = "302"; return dump; } catch (Exception e) { statusDescription = HttpClientMetricsService.unwrapExceptionString(e); throw e; } finally { httpMetrics.update(dump.url, statusDescription, System.currentTimeMillis() - before); } } }
BgpRisDownloader { public <T> BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper) { log.info("attempting to download new BGP RIS preview dump from {}", dump.url); long before = System.currentTimeMillis(); String statusDescription = "200"; final Supplier<Request> requestSupplier = () -> { final Request request = httpClient.newRequest(dump.url); if (dump.lastModified != null) { log.debug("Adding 'If-Modified-Since' equals to {}", formatAsRFC2616(dump.lastModified)); request.header("If-Modified-Since", formatAsRFC2616(dump.lastModified)); } return request; }; final BiFunction<InputStream, Long, BgpRisDump> streamReader = (stream, lastModified) -> { try { Stream<BgpRisEntry> entries = parse(new GZIPInputStream(stream)); ImmutableList.Builder<T> builder = ImmutableList.builder(); entries.flatMap(mapper).forEach(builder::add); return BgpRisDump.of(dump.url, new DateTime(lastModified), Optional.of(builder.build())); } catch (Exception e) { log.error("Error downloading RIS dump: " + dump.url); return dump; } }; try { return HttpStreaming.readStream(requestSupplier, streamReader); } catch (HttpStreaming.NotModifiedException n) { statusDescription = "302"; return dump; } catch (Exception e) { statusDescription = HttpClientMetricsService.unwrapExceptionString(e); throw e; } finally { httpMetrics.update(dump.url, statusDescription, System.currentTimeMillis() - before); } } @Autowired  BgpRisDownloader(HttpClientMetricsService httpMetrics, HttpClient httpClient); }
BgpRisDownloader { public <T> BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper) { log.info("attempting to download new BGP RIS preview dump from {}", dump.url); long before = System.currentTimeMillis(); String statusDescription = "200"; final Supplier<Request> requestSupplier = () -> { final Request request = httpClient.newRequest(dump.url); if (dump.lastModified != null) { log.debug("Adding 'If-Modified-Since' equals to {}", formatAsRFC2616(dump.lastModified)); request.header("If-Modified-Since", formatAsRFC2616(dump.lastModified)); } return request; }; final BiFunction<InputStream, Long, BgpRisDump> streamReader = (stream, lastModified) -> { try { Stream<BgpRisEntry> entries = parse(new GZIPInputStream(stream)); ImmutableList.Builder<T> builder = ImmutableList.builder(); entries.flatMap(mapper).forEach(builder::add); return BgpRisDump.of(dump.url, new DateTime(lastModified), Optional.of(builder.build())); } catch (Exception e) { log.error("Error downloading RIS dump: " + dump.url); return dump; } }; try { return HttpStreaming.readStream(requestSupplier, streamReader); } catch (HttpStreaming.NotModifiedException n) { statusDescription = "302"; return dump; } catch (Exception e) { statusDescription = HttpClientMetricsService.unwrapExceptionString(e); throw e; } finally { httpMetrics.update(dump.url, statusDescription, System.currentTimeMillis() - before); } } @Autowired  BgpRisDownloader(HttpClientMetricsService httpMetrics, HttpClient httpClient); BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper); static Stream<BgpRisEntry> parse(final InputStream is); }
BgpRisDownloader { public <T> BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper) { log.info("attempting to download new BGP RIS preview dump from {}", dump.url); long before = System.currentTimeMillis(); String statusDescription = "200"; final Supplier<Request> requestSupplier = () -> { final Request request = httpClient.newRequest(dump.url); if (dump.lastModified != null) { log.debug("Adding 'If-Modified-Since' equals to {}", formatAsRFC2616(dump.lastModified)); request.header("If-Modified-Since", formatAsRFC2616(dump.lastModified)); } return request; }; final BiFunction<InputStream, Long, BgpRisDump> streamReader = (stream, lastModified) -> { try { Stream<BgpRisEntry> entries = parse(new GZIPInputStream(stream)); ImmutableList.Builder<T> builder = ImmutableList.builder(); entries.flatMap(mapper).forEach(builder::add); return BgpRisDump.of(dump.url, new DateTime(lastModified), Optional.of(builder.build())); } catch (Exception e) { log.error("Error downloading RIS dump: " + dump.url); return dump; } }; try { return HttpStreaming.readStream(requestSupplier, streamReader); } catch (HttpStreaming.NotModifiedException n) { statusDescription = "302"; return dump; } catch (Exception e) { statusDescription = HttpClientMetricsService.unwrapExceptionString(e); throw e; } finally { httpMetrics.update(dump.url, statusDescription, System.currentTimeMillis() - before); } } @Autowired  BgpRisDownloader(HttpClientMetricsService httpMetrics, HttpClient httpClient); BgpRisDump<T> fetch(@NotNull BgpRisDump dump, Function<BgpRisEntry, Stream<T>> mapper); static Stream<BgpRisEntry> parse(final InputStream is);  }