StreamExecutionListener { public List<RowMetaAndData> getCachePreWindow() { return this.cachePreWindow; } StreamExecutionListener( final PublishSubject<RowMetaAndData> stream,
                                  Consumer<List<RowMetaAndData>> windowConsumer,
                                  final IDataServiceClientService.StreamingMode windowMode, final long windowSize,
                                  final long windowEvery, final int maxRows, final long maxTime ); List<RowMetaAndData> getCachePreWindow(); void unSubscribe();  }
@Test public void testGetCachedWindow() { streamExecutionListener = new StreamExecutionListener( streamList.getStream(), rowMetaAndDataList -> consumer.onNext( rowMetaAndDataList ) , IDataServiceClientService.StreamingMode.ROW_BASED, 10000, 1, 10000, 1000 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); streamList.add( mockRowMetaAndData ); assertFalse( streamExecutionListener.getCachePreWindow().isEmpty() ); assertEquals( 1, streamExecutionListener.getCachePreWindow().size() ); assertSame( mockRowMetaAndData, streamExecutionListener.getCachePreWindow().get( 0 ) ); streamList.add( mockRowMetaAndData2 ); assertFalse( streamExecutionListener.getCachePreWindow().isEmpty() ); assertEquals( 2, streamExecutionListener.getCachePreWindow().size() ); assertSame( mockRowMetaAndData, streamExecutionListener.getCachePreWindow().get( 0 ) ); assertSame( mockRowMetaAndData2, streamExecutionListener.getCachePreWindow().get( 1 ) ); }
@Test public void testGetCachedWindowTimeBased() throws Exception { streamExecutionListener = new StreamExecutionListener( streamList.getStream(), rowMetaAndDataList -> consumer.onNext( rowMetaAndDataList ) , IDataServiceClientService.StreamingMode.TIME_BASED, 10000,10, 1000, 100000 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); streamList.add( mockRowMetaAndData ); Thread.sleep( 100 ); assertFalse( streamExecutionListener.getCachePreWindow().isEmpty() ); assertEquals( 1, streamExecutionListener.getCachePreWindow().size() ); assertEquals( mockRowMetaAndData, streamExecutionListener.getCachePreWindow().get( 0 ) ); }
@Test public void testGetCachedWindowRegular() { streamExecutionListener = new StreamExecutionListener( streamList.getStream(), rowMetaAndDataList -> consumer.onNext( rowMetaAndDataList ) , IDataServiceClientService.StreamingMode.ROW_BASED, 2, 1, 10000, 50000 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); streamList.add( mockRowMetaAndData ); assertEquals( 1, listConsumer.size() ); streamList.add( mockRowMetaAndData2 ); assertEquals( 2, listConsumer.size() ); assertSame( mockRowMetaAndData, listConsumer.get( 0 ) ); assertSame( mockRowMetaAndData2, listConsumer.get( 1 ) ); }
@Test public void testGetCachedWindowRegularZeroEvery() { streamExecutionListener = new StreamExecutionListener( streamList.getStream(), rowMetaAndDataList -> consumer.onNext( rowMetaAndDataList ) , IDataServiceClientService.StreamingMode.ROW_BASED, 2, 0, 10000, 50000 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); streamList.add( mockRowMetaAndData ); assertEquals( 0, listConsumer.size() ); streamList.add( mockRowMetaAndData2 ); assertEquals( 2, listConsumer.size() ); assertSame( mockRowMetaAndData, listConsumer.get( 0 ) ); assertSame( mockRowMetaAndData2, listConsumer.get( 1 ) ); }
@Test public void testGetCachedWindowTimeBasedRegularZeroEvery() throws InterruptedException { streamExecutionListener = new StreamExecutionListener( streamList.getStream(), rowMetaAndDataList -> consumer.onNext( rowMetaAndDataList ) , IDataServiceClientService.StreamingMode.TIME_BASED, 200, 0, 10000, 50000 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); streamList.add( mockRowMetaAndData ); assertEquals( 0, listConsumer.size() ); streamList.add( mockRowMetaAndData2 ); assertEquals( 0, listConsumer.size() ); }