@Test public void groupOperations_WithThreeTimesGroupMaxSizeElements_EmitsThreeGroups() { contentStoreCore.groupOperations(createSource(3 * DEFAULT_GROUP_MAX_SIZE)) .test() .awaitDone(2 * DEFAULT_GROUPING_TIMEOUT_MS, TimeUnit.MILLISECONDS) .assertNotComplete() .assertValueCount(3); }
@NonNull protected <R> Observable<List<R>> groupOperations(@NonNull final Observable<R> source) { return source.publish(stream -> stream.buffer( Observable.merge( stream.window(groupMaxSize).skip(1), stream.debounce(groupingTimeout, TimeUnit.MILLISECONDS)))) .filter(list -> !list.isEmpty()); }
ContentProviderStoreCoreBase { @NonNull protected <R> Observable<List<R>> groupOperations(@NonNull final Observable<R> source) { return source.publish(stream -> stream.buffer( Observable.merge( stream.window(groupMaxSize).skip(1), stream.debounce(groupingTimeout, TimeUnit.MILLISECONDS)))) .filter(list -> !list.isEmpty()); } }
ContentProviderStoreCoreBase { @NonNull protected <R> Observable<List<R>> groupOperations(@NonNull final Observable<R> source) { return source.publish(stream -> stream.buffer( Observable.merge( stream.window(groupMaxSize).skip(1), stream.debounce(groupingTimeout, TimeUnit.MILLISECONDS)))) .filter(list -> !list.isEmpty()); } protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver); protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver,
                                           final int groupingTimeout,
                                           final int groupMaxSize); }
ContentProviderStoreCoreBase { @NonNull protected <R> Observable<List<R>> groupOperations(@NonNull final Observable<R> source) { return source.publish(stream -> stream.buffer( Observable.merge( stream.window(groupMaxSize).skip(1), stream.debounce(groupingTimeout, TimeUnit.MILLISECONDS)))) .filter(list -> !list.isEmpty()); } protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver); protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver,
                                           final int groupingTimeout,
                                           final int groupMaxSize); @NonNull static Handler createHandler(@NonNull final String name); }
ContentProviderStoreCoreBase { @NonNull protected <R> Observable<List<R>> groupOperations(@NonNull final Observable<R> source) { return source.publish(stream -> stream.buffer( Observable.merge( stream.window(groupMaxSize).skip(1), stream.debounce(groupingTimeout, TimeUnit.MILLISECONDS)))) .filter(list -> !list.isEmpty()); } protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver); protected  ContentProviderStoreCoreBase(@NonNull final ContentResolver contentResolver,
                                           final int groupingTimeout,
                                           final int groupMaxSize); @NonNull static Handler createHandler(@NonNull final String name);  }