1515
1616package utils ;
1717
18+ import io .reactivex .Flowable ;
1819import org .junit .Before ;
1920import org .junit .Test ;
2021import org .junit .runner .RunWith ;
2122import org .mockito .Mock ;
2223import org .mockito .Mockito ;
2324import org .mockito .junit .MockitoJUnitRunner ;
2425import org .reactivestreams .Subscriber ;
25- import software .amazon .awssdk .core .pagination .async .AsyncPageFetcher ;
26- import software .amazon .awssdk .core .pagination .async .PaginatedItemsPublisher ;
26+ import software .amazon .awssdk .core .async .SdkPublisher ;
2727import software .amazon .awssdk .utils .async .LimitingSubscriber ;
2828import software .amazon .awssdk .utils .internal .async .EmptySubscription ;
2929
3030import java .util .ArrayList ;
3131import java .util .Arrays ;
3232import java .util .Iterator ;
3333import java .util .List ;
34- import java .util .concurrent .CompletableFuture ;
3534import java .util .concurrent .ExecutionException ;
3635import java .util .concurrent .ExecutorService ;
3736import java .util .concurrent .Executors ;
@@ -48,96 +47,83 @@ public class SdkSubscriberTest {
4847
4948 public static final Function <Integer , Iterator <Integer >> SAMPLE_ITERATOR = response -> Arrays .asList (1 , 2 , 3 , 4 , 5 , 6 ).listIterator ();
5049 public static final Function <Integer , Iterator <Integer >> EMPTY_ITERATOR = response -> new ArrayList <Integer >().listIterator ();
51- @ Mock
52- AsyncPageFetcher asyncPageFetcher ;
53- PaginatedItemsPublisher <Integer , Integer > itemsPublisher ;
5450
5551 @ Mock
5652 Subscriber <Integer > mockSubscriber ;
5753
54+ private SdkPublisher <Integer > sdkPublisher ;
55+
5856 @ Before
5957 public void setUp () {
60- doReturn (CompletableFuture .completedFuture (1 ))
61- .when (asyncPageFetcher ).nextPage (null );
62- doReturn (false )
63- .when (asyncPageFetcher ).hasNextPage (any ());
58+ sdkPublisher = SdkPublisher .adapt (Flowable .just (1 , 2 , 3 , 4 , 5 , 6 ));
6459 }
6560
6661 @ Test
6762 public void limitingSubscriber_with_different_limits () throws InterruptedException , ExecutionException , TimeoutException {
68- itemsPublisher = PaginatedItemsPublisher .builder ().nextPageFetcher (asyncPageFetcher )
69- .iteratorFunction (SAMPLE_ITERATOR ).isLastPage (false ).build ();
63+ List <Integer > belowLimit = new ArrayList <>();
64+ sdkPublisher .limit (3 ).subscribe (belowLimit ::add ).get (5 , TimeUnit .SECONDS );
65+ assertThat (belowLimit ).containsExactly (1 , 2 , 3 );
7066
71- final List <Integer > belowLimit = new ArrayList <>();
72- itemsPublisher .limit (3 ).subscribe (e -> belowLimit . add ( e ) ).get (5 , TimeUnit .SECONDS );
73- assertThat (belowLimit ). isEqualTo ( Arrays . asList ( 1 , 2 , 3 ) );
67+ List <Integer > beyondLimit = new ArrayList <>();
68+ sdkPublisher .limit (33 ).subscribe (beyondLimit :: add ).get (5 , TimeUnit .SECONDS );
69+ assertThat (beyondLimit ). containsExactly ( 1 , 2 , 3 , 4 , 5 , 6 );
7470
75- final List <Integer > beyondLimit = new ArrayList <>();
76- itemsPublisher .limit (33 ).subscribe (e -> beyondLimit .add (e )).get (5 , TimeUnit .SECONDS );
77- assertThat (beyondLimit ).isEqualTo (Arrays .asList (1 , 2 , 3 , 4 , 5 , 6 ));
78-
79- final List <Integer > zeroLimit = new ArrayList <>();
80- itemsPublisher .limit (0 ).subscribe (e -> zeroLimit .add (e )).get (5 , TimeUnit .SECONDS );
81- assertThat (zeroLimit ).isEqualTo (Arrays .asList ());
71+ List <Integer > zeroLimit = new ArrayList <>();
72+ sdkPublisher .limit (0 ).subscribe (zeroLimit ::add ).get (5 , TimeUnit .SECONDS );
73+ assertThat (zeroLimit ).isEmpty ();
8274 }
8375
8476 @ Test
8577 public void filteringSubscriber_with_different_filters () throws InterruptedException , ExecutionException , TimeoutException {
86- itemsPublisher = PaginatedItemsPublisher .builder ().nextPageFetcher (asyncPageFetcher )
87- .iteratorFunction (SAMPLE_ITERATOR ).isLastPage (false ).build ();
88-
89- final List <Integer > filteredSomeList = new ArrayList <>();
90- itemsPublisher .filter (i -> i % 2 == 0 ).subscribe (e -> filteredSomeList .add (e )).get (5 , TimeUnit .SECONDS );
91- assertThat (filteredSomeList ).isEqualTo (Arrays .asList (2 , 4 , 6 ));
9278
93- final List <Integer > filteredAllList = new ArrayList <>();
94- itemsPublisher .filter (i -> i % 10 == 0 ).subscribe (e -> filteredAllList . add ( e ) ).get (5 , TimeUnit .SECONDS );
95- assertThat (filteredAllList ). isEqualTo ( Arrays . asList () );
79+ List <Integer > filteredSomeList = new ArrayList <>();
80+ sdkPublisher .filter (i -> i % 2 == 0 ).subscribe (filteredSomeList :: add ).get (5 , TimeUnit .SECONDS );
81+ assertThat (filteredSomeList ). containsExactly ( 2 , 4 , 6 );
9682
97- final List <Integer > filteredNone = new ArrayList <>();
98- itemsPublisher .filter (i -> i % 1 == 0 ).subscribe (e -> filteredNone . add ( e ) ).get (5 , TimeUnit .SECONDS );
99- assertThat (filteredNone ). isEqualTo ( Arrays . asList ( 1 , 2 , 3 , 4 , 5 , 6 ) );
83+ List <Integer > filteredAllList = new ArrayList <>();
84+ sdkPublisher .filter (i -> i % 10 == 0 ).subscribe (filteredAllList :: add ).get (5 , TimeUnit .SECONDS );
85+ assertThat (filteredAllList ). isEmpty ( );
10086
87+ List <Integer > filteredNone = new ArrayList <>();
88+ sdkPublisher .filter (i -> i % 1 == 0 ).subscribe (filteredNone ::add ).get (5 , TimeUnit .SECONDS );
89+ assertThat (filteredNone ).containsExactly (1 , 2 , 3 , 4 , 5 , 6 );
10190 }
10291
10392 @ Test
10493 public void limit_and_filter_subscriber_chained_with_different_conditions () throws InterruptedException , ExecutionException , TimeoutException {
105- itemsPublisher = PaginatedItemsPublisher .builder ().nextPageFetcher (asyncPageFetcher )
106- .iteratorFunction (SAMPLE_ITERATOR ).isLastPage (false ).build ();
10794
108- final List <Integer > belowLimitWithFiltering = new ArrayList <>();
109- itemsPublisher .limit (4 ).filter (i -> i % 2 == 0 ).subscribe (e -> belowLimitWithFiltering . add ( e ) ).get (5 , TimeUnit .SECONDS );
110- assertThat (belowLimitWithFiltering ).isEqualTo ( Arrays . asList ( 2 , 4 ) );
95+ List <Integer > belowLimitWithFiltering = new ArrayList <>();
96+ sdkPublisher .limit (4 ).filter (i -> i % 2 == 0 ).subscribe (belowLimitWithFiltering :: add ).get (5 , TimeUnit .SECONDS );
97+ assertThat (belowLimitWithFiltering ).containsExactly ( 2 , 4 );
11198
112- final List <Integer > beyondLimitWithAllFiltering = new ArrayList <>();
113- itemsPublisher .limit (33 ).filter (i -> i % 10 == 0 ).subscribe (e -> beyondLimitWithAllFiltering . add ( e ) ).get (5 , TimeUnit .SECONDS );
114- assertThat (beyondLimitWithAllFiltering ).isEqualTo ( Arrays . asList () );
99+ List <Integer > beyondLimitWithAllFiltering = new ArrayList <>();
100+ sdkPublisher .limit (33 ).filter (i -> i % 10 == 0 ).subscribe (beyondLimitWithAllFiltering :: add ).get (5 , TimeUnit .SECONDS );
101+ assertThat (beyondLimitWithAllFiltering ).isEmpty ( );
115102
116- final List <Integer > zeroLimitAndNoFilter = new ArrayList <>();
117- itemsPublisher .limit (0 ).filter (i -> i % 1 == 0 ).subscribe (e -> zeroLimitAndNoFilter . add ( e ) ).get (5 , TimeUnit .SECONDS );
118- assertThat (zeroLimitAndNoFilter ).isEqualTo ( Arrays . asList () );
103+ List <Integer > zeroLimitAndNoFilter = new ArrayList <>();
104+ sdkPublisher .limit (0 ).filter (i -> i % 1 == 0 ).subscribe (zeroLimitAndNoFilter :: add ).get (5 , TimeUnit .SECONDS );
105+ assertThat (zeroLimitAndNoFilter ).isEmpty ( );
119106
120- final List <Integer > filteringbelowLimitWith = new ArrayList <>();
121- itemsPublisher .filter (i -> i % 2 == 0 ).limit (2 ).subscribe (e -> filteringbelowLimitWith . add ( e ) ).get (5 , TimeUnit .SECONDS );
122- assertThat (filteringbelowLimitWith ).isEqualTo ( Arrays . asList ( 2 , 4 ) );
107+ List <Integer > filteringbelowLimitWith = new ArrayList <>();
108+ sdkPublisher .filter (i -> i % 2 == 0 ).limit (2 ).subscribe (filteringbelowLimitWith :: add ).get (5 , TimeUnit .SECONDS );
109+ assertThat (filteringbelowLimitWith ).containsExactly ( 2 , 4 );
123110
124- final List <Integer > filteringAndOutsideLimit = new ArrayList <>();
125- itemsPublisher .filter (i -> i % 10 == 0 ).limit (33 ).subscribe (e -> filteringAndOutsideLimit . add ( e ) ).get (5 , TimeUnit .SECONDS );
126- assertThat (filteringAndOutsideLimit ).isEqualTo ( Arrays . asList () );
111+ List <Integer > filteringAndOutsideLimit = new ArrayList <>();
112+ sdkPublisher .filter (i -> i % 10 == 0 ).limit (33 ).subscribe (filteringAndOutsideLimit :: add ).get (5 , TimeUnit .SECONDS );
113+ assertThat (filteringAndOutsideLimit ).isEmpty ( );
127114 }
128115
129116 @ Test
130117 public void limit__subscriber_with_empty_input_and_zero_limit () throws Exception {
131- itemsPublisher = PaginatedItemsPublisher .builder ().nextPageFetcher (asyncPageFetcher )
132- .iteratorFunction (EMPTY_ITERATOR ).isLastPage (false ).build ();
118+ sdkPublisher = SdkPublisher .adapt (Flowable .empty ());
133119
134- final List <Integer > zeroLimit = new ArrayList <>();
135- itemsPublisher .limit (0 ).subscribe (e -> zeroLimit . add ( e ) ).get (5 , TimeUnit .SECONDS );
136- assertThat (zeroLimit ).isEqualTo ( Arrays . asList () );
120+ List <Integer > zeroLimit = new ArrayList <>();
121+ sdkPublisher .limit (0 ).subscribe (zeroLimit :: add ).get (5 , TimeUnit .SECONDS );
122+ assertThat (zeroLimit ).isEmpty ( );
137123
138124 List <Integer > nonZeroLimit = new ArrayList <>();
139- itemsPublisher .limit (10 ).subscribe (e -> nonZeroLimit . add ( e ) ).get (5 , TimeUnit .SECONDS );
140- assertThat (zeroLimit ).isEqualTo ( Arrays . asList () );
125+ sdkPublisher .limit (10 ).subscribe (nonZeroLimit :: add ).get (5 , TimeUnit .SECONDS );
126+ assertThat (zeroLimit ).isEmpty ( );
141127 }
142128
143129
0 commit comments