@Test public void checkpointWithNoScope() { NakadiClient client = spy(NakadiClient.newBuilder() .baseURI("http: .tokenProvider(scope -> { if (scope != null) { throw new AssertionError("scope should not be called"); } return Optional.empty(); }) .build()); Resource r0 = spy(new OkHttpResource( new OkHttpClient.Builder().build(), new GsonSupport(), mock(MetricCollector.class))); when(client.resourceProvider()).thenReturn(mock(ResourceProvider.class)); when(client.resourceProvider().newResource()).thenReturn(r0); Cursor c = new Cursor().cursorToken("ctoken").eventType("et1").offset("0").partition("1"); Map<String, Object> requestMap = Maps.newHashMap(); requestMap.put("items", Lists.newArrayList(c)); HashMap<String, String> context = Maps.newHashMap(); context.put("X-Nakadi-StreamId", "aa"); context.put("SubscriptionId", "woo"); try { RetryPolicy backoff = ExponentialRetry.newBuilder() .initialInterval(1, TimeUnit.SECONDS) .maxAttempts(1) .maxInterval(1, TimeUnit.SECONDS) .build(); new SubscriptionResourceReal(client) .checkpoint(backoff, context, c); } catch (RetryableException | NotFoundException ignored) { } ArgumentCaptor<ResourceOptions> options = ArgumentCaptor.forClass(ResourceOptions.class); verify(r0, times(1)).requestThrowing( Matchers.eq(Resource.POST), Matchers.eq("http: options.capture(), Matchers.any(ContentSupplier.class) ); Resource r1 = spy(new OkHttpResource( new OkHttpClient.Builder().build(), new GsonSupport(), mock(MetricCollector.class))); when(client.resourceProvider()).thenReturn(mock(ResourceProvider.class)); when(client.resourceProvider().newResource()).thenReturn(r1); try { RetryPolicy backoff = ExponentialRetry.newBuilder() .initialInterval(1, TimeUnit.SECONDS) .maxAttempts(1) .maxInterval(1, TimeUnit.SECONDS) .build(); SubscriptionResource resource = new SubscriptionResourceReal(client); ((SubscriptionResourceReal)resource).checkpoint(backoff, context, c); } catch (RetryableException | NotFoundException ignored) { } options = ArgumentCaptor.forClass(ResourceOptions.class); verify(r1, times(1)).requestThrowing( Matchers.eq(Resource.POST), Matchers.eq("http: options.capture(), Matchers.any(ContentSupplier.class) ); assertEquals(null, options.getValue().scope()); }
@Override public Response requestThrowing(String method, String url, ResourceOptions options) throws NakadiException { return maybeComposeRetryPolicy( Observable.defer( () -> Observable.just(requestThrowingInner(method, url, options)))).blockingFirst(); }
OkHttpResource implements Resource { @Override public Response requestThrowing(String method, String url, ResourceOptions options) throws NakadiException { return maybeComposeRetryPolicy( Observable.defer( () -> Observable.just(requestThrowingInner(method, url, options)))).blockingFirst(); } }
OkHttpResource implements Resource { @Override public Response requestThrowing(String method, String url, ResourceOptions options) throws NakadiException { return maybeComposeRetryPolicy( Observable.defer( () -> Observable.just(requestThrowingInner(method, url, options)))).blockingFirst(); } OkHttpResource(OkHttpClient okHttpClient, JsonSupport jsonSupport, MetricCollector collector); }
OkHttpResource implements Resource { @Override public Response requestThrowing(String method, String url, ResourceOptions options) throws NakadiException { return maybeComposeRetryPolicy( Observable.defer( () -> Observable.just(requestThrowingInner(method, url, options)))).blockingFirst(); } OkHttpResource(OkHttpClient okHttpClient, JsonSupport jsonSupport, MetricCollector collector); OkHttpResource connectTimeout(long timeout, TimeUnit unit); OkHttpResource readTimeout(long timeout, TimeUnit unit); @Override Resource writeTimeout(long timeout, TimeUnit unit); OkHttpResource retryPolicy(RetryPolicy retryPolicy); @Override Response request(String method, String url, ResourceOptions options); @Override Response requestThrowing(String method, String url, ResourceOptions options); @Override Response requestThrowing(String method, String url, ResourceOptions options,
      ContentSupplier body); @Override Response postEventsThrowing(String url, ResourceOptions options, ContentSupplier body); @Override Res requestThrowing(String method, String url, ResourceOptions options,
      Class<Res> res); @Override Res requestThrowing(String method, String url, ResourceOptions options,
      ContentSupplier body, Class<Res> res); }
OkHttpResource implements Resource { @Override public Response requestThrowing(String method, String url, ResourceOptions options) throws NakadiException { return maybeComposeRetryPolicy( Observable.defer( () -> Observable.just(requestThrowingInner(method, url, options)))).blockingFirst(); } OkHttpResource(OkHttpClient okHttpClient, JsonSupport jsonSupport, MetricCollector collector); OkHttpResource connectTimeout(long timeout, TimeUnit unit); OkHttpResource readTimeout(long timeout, TimeUnit unit); @Override Resource writeTimeout(long timeout, TimeUnit unit); OkHttpResource retryPolicy(RetryPolicy retryPolicy); @Override Response request(String method, String url, ResourceOptions options); @Override Response requestThrowing(String method, String url, ResourceOptions options); @Override Response requestThrowing(String method, String url, ResourceOptions options,
      ContentSupplier body); @Override Response postEventsThrowing(String url, ResourceOptions options, ContentSupplier body); @Override Res requestThrowing(String method, String url, ResourceOptions options,
      Class<Res> res); @Override Res requestThrowing(String method, String url, ResourceOptions options,
      ContentSupplier body, Class<Res> res);  }