StreamExecutionListener { public void unSubscribe() { unSubscribeOutput(); unSubscribeStarter(); unSubscribeBuffer(); unSubscribeFallbackBuffer(); } 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 testUnsubscribe() { 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 ) ); streamExecutionListener.unSubscribe(); streamList.add( mockRowMetaAndData2 ); assertTrue( streamExecutionListener.getCachePreWindow().isEmpty() ); assertEquals( 1, listConsumer.size() ); assertSame( mockRowMetaAndData, listConsumer.get( 0 ) ); }