@Test void cancelRequestOnceUnsubscribed() { final AtomicReference<BucketConfigRequest> request = new AtomicReference<>(); doAnswer(i -> { request.set(i.getArgument(0)); return null; }).when(core).send(any(BucketConfigRequest.class)); Disposable disposable = loader.discoverConfig(SEED, BUCKET).subscribe(); disposable.dispose(); assertTrue(request.get().completed()); assertTrue(request.get().cancelled()); assertEquals(CancellationReason.STOPPED_LISTENING, request.get().cancellationReason()); }
@Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); }
ClusterManagerBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); } }
ClusterManagerBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); } ClusterManagerBucketLoader(final Core core); }
ClusterManagerBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); } ClusterManagerBucketLoader(final Core core);  }
ClusterManagerBucketLoader extends BaseBucketLoader { @Override protected Mono<byte[]> discoverConfig(final NodeIdentifier seed, final String bucket) { final CoreContext ctx = core().context(); return Mono.defer(() -> { BucketConfigRequest request = new BucketConfigRequest( ctx.environment().timeoutConfig().connectTimeout(), ctx, BestEffortRetryStrategy.INSTANCE, bucket, ctx.authenticator(), seed ); core().send(request); return Reactor.wrap(request, request.response(), true); }).map(response -> { if (response.status().success()) { return response.config(); } else { throw new ConfigException("Received error status from ClusterManagerBucketLoader: " + response); } }); } ClusterManagerBucketLoader(final Core core);   }