JCBC-1209: Audit and add explicit subscribers on req creation

Motivation
----------
In an earlier changeset the explicit subscriber management on
the request has been added, but some of the APIs have been
overlooked - this can lead to inconsistent behavior
in core-io.

Modifications
-------------
This changeset audits all calls to core.send(...) and where
needed adds the explicit subscriber management on the request
since core-io checks if a subscriber is attached and if not
it never treats it as non-active. Especially n1ql requests
in this regard have been treaded differently before this
changeset compared to kv requests.

Result
------
Complete audit of .send() APIs and every request now properly
handles it subscriber and can be checked against timeouts in
core-io.

Change-Id: I71f752384af7ac4fa88534531401901f92a1fb24
Reviewed-on: http://review.couchbase.org/95227
Reviewed-by: Michael Nitschinger <[email protected]>
Tested-by: Michael Nitschinger <[email protected]>

Author: daschl

src/main/java/com/couchbase/client/java/CouchbaseAsyncBucket.java

@@ -813,11 +813,12 @@ public Observable<? extends D> call(Throwable throwable) {
 
     @Override
     public Observable<AsyncViewResult> query(final ViewQuery query) {
-        Observable<ViewQueryResponse> source = Observable.defer(new Func0<Observable<ViewQueryResponse>>() {
+        Observable<ViewQueryResponse> source = deferAndWatch(new Func1<Subscriber, Observable<ViewQueryResponse>>() {
             @Override
-            public Observable<ViewQueryResponse> call() {
+            public Observable<ViewQueryResponse> call(Subscriber subscriber) {
                 final ViewQueryRequest request = new ViewQueryRequest(query.getDesign(), query.getView(),
                     query.isDevelopment(), query.toQueryString(), query.getKeys(), bucket, username, password);
+                request.subscriber(subscriber);
                 return core.send(request);
             }
         });
@@ -841,11 +842,12 @@ public Observable<AsyncSearchQueryResult> query(final SearchQuery query) {
             query.serverSideTimeout(environment().searchTimeout(), TimeUnit.MILLISECONDS);
         }
 
-        Observable<SearchQueryResponse> source = Observable.defer(new Func0<Observable<SearchQueryResponse>>() {
+        Observable<SearchQueryResponse> source = deferAndWatch(new Func1<Subscriber, Observable<SearchQueryResponse>>() {
             @Override
-            public Observable<SearchQueryResponse> call() {
+            public Observable<SearchQueryResponse> call(Subscriber subscriber) {
                 final SearchQueryRequest request =
-                    new SearchQueryRequest(indexName, query.export().toString(), bucket, username, password);
+                        new SearchQueryRequest(indexName, query.export().toString(), bucket, username, password);
+                request.subscriber(subscriber);
                 return core.send(request);
             }
         });
@@ -872,11 +874,12 @@ public AsyncSearchQueryResult call(SearchQueryResponse response) {
 
     @Override
     public Observable<AsyncSpatialViewResult> query(final SpatialViewQuery query) {
-        Observable<ViewQueryResponse> source = Observable.defer(new Func0<Observable<ViewQueryResponse>>() {
+        Observable<ViewQueryResponse> source = deferAndWatch(new Func1<Subscriber, Observable<ViewQueryResponse>>() {
             @Override
-            public Observable<ViewQueryResponse> call() {
+            public Observable<ViewQueryResponse> call(Subscriber subscriber) {
                 final ViewQueryRequest request = new ViewQueryRequest(query.getDesign(), query.getView(),
-                    query.isDevelopment(), true, query.toString(), null, bucket, username, password);
+                        query.isDevelopment(), true, query.toString(), null, bucket, username, password);
+                request.subscriber(subscriber);
                 return core.send(request);
             }
         });

src/main/java/com/couchbase/client/java/analytics/AnalyticsQueryExecutor.java

@@ -23,14 +23,15 @@
 import com.couchbase.client.java.error.TranscodingException;
 import com.couchbase.client.java.transcoder.TranscoderUtils;
 import rx.Observable;
-import rx.functions.Func0;
+import rx.Subscriber;
 import rx.functions.Func1;
 import rx.functions.Func6;
 
 import java.util.Arrays;
 import java.util.List;
 
 import static com.couchbase.client.java.CouchbaseAsyncBucket.JSON_OBJECT_TRANSCODER;
+import static com.couchbase.client.java.util.OnSubscribeDeferAndWatch.deferAndWatch;
 
 public class AnalyticsQueryExecutor {
 
@@ -47,10 +48,13 @@ public AnalyticsQueryExecutor(ClusterFacade core, String bucket, String username
     }
 
     public Observable<AsyncAnalyticsQueryResult> execute(final AnalyticsQuery query) {
-        return Observable.defer(new Func0<Observable<GenericAnalyticsResponse>>() {
+        return deferAndWatch(new Func1<Subscriber, Observable<GenericAnalyticsResponse>>() {
             @Override
-            public Observable<GenericAnalyticsResponse> call() {
-                return core.send(GenericAnalyticsRequest.jsonQuery(query.query().toString(), bucket, username, password));
+            public Observable<GenericAnalyticsResponse> call(final Subscriber subscriber) {
+                GenericAnalyticsRequest request = GenericAnalyticsRequest
+                        .jsonQuery(query.query().toString(), bucket, username, password);
+                request.subscriber(subscriber);
+                return core.<GenericAnalyticsResponse>send(request);
             }
         }).flatMap(new Func1<GenericAnalyticsResponse, Observable<AsyncAnalyticsQueryResult>>() {
             @Override

src/main/java/com/couchbase/client/java/bucket/BucketFlusher.java

@@ -31,6 +31,7 @@
 import com.couchbase.client.deps.io.netty.util.CharsetUtil;
 import com.couchbase.client.java.error.FlushDisabledException;
 import rx.Observable;
+import rx.Subscriber;
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Func2;
@@ -39,6 +40,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
+import static com.couchbase.client.java.util.OnSubscribeDeferAndWatch.deferAndWatch;
 import static com.couchbase.client.java.util.retry.RetryBuilder.any;
 
 /**
@@ -118,8 +120,15 @@ private static Observable<List<String>> createMarkerDocuments(final ClusterFacad
             .from(FLUSH_MARKERS)
             .flatMap(new Func1<String, Observable<UpsertResponse>>() {
                 @Override
-                public Observable<UpsertResponse> call(String id) {
-                    return core.send(new UpsertRequest(id, Unpooled.copiedBuffer(id, CharsetUtil.UTF_8), bucket));
+                public Observable<UpsertResponse> call(final String id) {
+                    return deferAndWatch(new Func1<Subscriber, Observable<? extends UpsertResponse>>() {
+                        @Override
+                        public Observable<? extends UpsertResponse> call(final Subscriber subscriber) {
+                            UpsertRequest request = new UpsertRequest(id, Unpooled.copiedBuffer(id, CharsetUtil.UTF_8), bucket);
+                            request.subscriber(subscriber);
+                            return core.send(request);
+                        }
+                    });
                 }
             })
             .doOnNext(new Action1<UpsertResponse>() {
@@ -152,8 +161,14 @@ public List<String> call(UpsertResponse response) {
      * @return an observable indicating if done (true) or polling needs to happen (false).
      */
     private static Observable<Boolean> initiateFlush(final ClusterFacade core, final String bucket, final String username, final String password) {
-        return core
-            .<FlushResponse>send(new FlushRequest(bucket, username, password))
+        return deferAndWatch(new Func1<Subscriber, Observable<FlushResponse>>() {
+                @Override
+                public Observable<FlushResponse> call(Subscriber subscriber) {
+                    FlushRequest request = new FlushRequest(bucket, username, password);
+                    request.subscriber(subscriber);
+                    return core.send(request);
+                }
+            })
             .retryWhen(any().delay(Delay.fixed(100, TimeUnit.MILLISECONDS)).max(Integer.MAX_VALUE).build())
             .map(new Func1<FlushResponse, Boolean>() {
                 @Override
@@ -182,8 +197,15 @@ private static Observable<Boolean> pollMarkerDocuments(final ClusterFacade core,
             .from(FLUSH_MARKERS)
             .flatMap(new Func1<String, Observable<GetResponse>>() {
                 @Override
-                public Observable<GetResponse> call(String id) {
-                    return core.send(new GetRequest(id, bucket));
+                public Observable<GetResponse> call(final String id) {
+                    return deferAndWatch(new Func1<Subscriber, Observable<? extends GetResponse>>() {
+                        @Override
+                        public Observable<? extends GetResponse> call(Subscriber subscriber) {
+                            GetRequest request = new GetRequest(id, bucket);
+                            request.subscriber(subscriber);
+                            return core.send(request);
+                        }
+                    });
                 }
             })
             .reduce(0, new Func2<Integer, GetResponse, Integer>() {

src/main/java/com/couchbase/client/java/bucket/DefaultAsyncBucketManager.java

@@ -117,7 +117,15 @@ public Observable<BucketInfo> info() {
         return Observable.defer(new Func0<Observable<BucketConfigResponse>>() {
             @Override
             public Observable<BucketConfigResponse> call() {
-                return core.send(new BucketConfigRequest("/pools/default/buckets/", null, bucket, username, password));
+                return deferAndWatch(new Func1<Subscriber, Observable<? extends BucketConfigResponse>>() {
+                    @Override
+                    public Observable<? extends BucketConfigResponse> call(Subscriber subscriber) {
+                        BucketConfigRequest request = new BucketConfigRequest("/pools/default/buckets/",
+                            null, bucket, username, password);
+                        request.subscriber(subscriber);
+                        return core.send(request);
+                    }
+                });
             }
         })
         .retryWhen(any().delay(Delay.fixed(100, TimeUnit.MILLISECONDS)).max(Integer.MAX_VALUE).build())
@@ -151,7 +159,14 @@ public Observable<DesignDocument> getDesignDocuments(final boolean development)
         return Observable.defer(new Func0<Observable<GetDesignDocumentsResponse>>() {
             @Override
             public Observable<GetDesignDocumentsResponse> call() {
-                return core.send(new GetDesignDocumentsRequest(bucket, username, password));
+                return deferAndWatch(new Func1<Subscriber, Observable<? extends GetDesignDocumentsResponse>>() {
+                    @Override
+                    public Observable<? extends GetDesignDocumentsResponse> call(final Subscriber subscriber) {
+                        GetDesignDocumentsRequest request = new GetDesignDocumentsRequest(bucket, username, password);
+                        request.subscriber(subscriber);
+                        return core.send(request);
+                    }
+                });
             }
         })
         .retryWhen(any().delay(Delay.fixed(100, TimeUnit.MILLISECONDS)).max(Integer.MAX_VALUE).build())

src/main/java/com/couchbase/client/java/bucket/ReplicaReader.java

@@ -33,11 +33,13 @@
 import com.couchbase.client.java.error.CouchbaseOutOfMemoryException;
 import com.couchbase.client.java.error.TemporaryFailureException;
 import rx.Observable;
+import rx.Subscriber;
 import rx.functions.Func0;
 import rx.functions.Func1;
 
 import java.util.ArrayList;
 import java.util.List;
+import static com.couchbase.client.java.util.OnSubscribeDeferAndWatch.deferAndWatch;
 
 /**
  * Helper class to deal with reading from zero to N replicas and returning results.
@@ -70,15 +72,19 @@ private ReplicaReader() {}
     public static Observable<GetResponse> read(final ClusterFacade core, final String id,
         final ReplicaMode type, final String bucket) {
         return assembleRequests(core, id, type, bucket)
-            .flatMap(new Func1<BinaryRequest, Observable<GetResponse>>() {
-                @Override
-                public Observable<GetResponse> call(BinaryRequest request) {
-                    return core
-                        .<GetResponse>send(request)
-                        .filter(GetResponseFilter.INSTANCE)
-                        .onErrorResumeNext(GetResponseErrorHandler.INSTANCE);
-                }
-            });
+                .flatMap(new Func1<BinaryRequest, Observable<GetResponse>>() {
+                    @Override
+                    public Observable<GetResponse> call(final BinaryRequest request) {
+                        Observable<GetResponse> result = deferAndWatch(new Func1<Subscriber, Observable<GetResponse>>() {
+                            @Override
+                            public Observable<GetResponse> call(Subscriber subscriber) {
+                                request.subscriber(subscriber);
+                                return core.send(request);
+                            }
+                        }).filter(GetResponseFilter.INSTANCE);
+                        return result.onErrorResumeNext(GetResponseErrorHandler.INSTANCE);
+                    }
+                });
     }
 
     /**