StreamingServiceTransExecutor { public StreamExecutionListener getBuffer( String query, Consumer<List<RowMetaAndData>> windowConsumer, final IDataServiceClientService.StreamingMode windowMode, long windowSize, long windowEvery, long windowLimit ) { String cacheId = WindowParametersHelper.getCacheKey( query, windowMode, windowSize, windowEvery, windowMaxRowLimit, windowMaxTimeLimit, windowLimit, getKey().hashCode() ); if ( cacheId == null ) { return null; } StreamExecutionListener streamListener = serviceListeners.getIfPresent( cacheId ); if ( streamListener == null ) { if ( stepStream == null ) { stepStream = new StreamList<>(); } boolean timeBased = IDataServiceClientService.StreamingMode.TIME_BASED.equals( windowMode ); boolean rowBased = IDataServiceClientService.StreamingMode.ROW_BASED.equals( windowMode ); int maxRows = WindowParametersHelper.getMaxRows( windowMaxRowLimit, windowLimit, timeBased ); long maxTime = WindowParametersHelper.getMaxTime( windowMaxTimeLimit, windowLimit, rowBased ); windowSize = WindowParametersHelper.getWindowSize( windowSize, timeBased, maxRows, maxTime ); windowEvery = WindowParametersHelper.getWindowEvery( windowEvery, timeBased, maxRows, maxTime ); streamListener = new StreamExecutionListener( stepStream.getStream(), windowConsumer, windowMode, windowSize, windowEvery, maxRows, maxTime ); serviceListeners.put( cacheId, streamListener ); } if ( isRunning.compareAndSet( false, true ) ) { startService(); } return streamListener; } StreamingServiceTransExecutor( final StreamServiceKey key, Trans serviceTrans, final String serviceStepName,
                                        final int windowMaxRowLimit, final long windowMaxTimeLimit,
                                        final Context context ); Trans getServiceTrans(); StreamServiceKey getKey(); long getWindowMaxRowLimit(); long getWindowMaxTimeLimit(); StreamExecutionListener getBuffer( String query, Consumer<List<RowMetaAndData>> windowConsumer, final IDataServiceClientService.StreamingMode windowMode,
                                            long windowSize, long windowEvery, long windowLimit ); void clearCache(); void clearCacheByKey( String key ); void stopAll(); StreamExecutionListener touchServiceListener( String key );  }
@Test ( expected = RuntimeException.class ) public void testServiceStepNotFound() { when( serviceTrans.findRunThread( MOCK_SERVICE_STEP_NAME ) ).thenReturn( null ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 0 ); }
@Test ( expected = RuntimeException.class ) public void testServiceStartThreadException() throws KettleException { doThrow( new KettleException() ).when( serviceTrans ).startThreads(); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 0 ); }
@Test public void testGetBufferWindowNullInvalidValues() throws Exception { StreamExecutionListener buffer = serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 0, 0, 0 ); assertNull( buffer ); buffer = serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, -1, 0, 0 ); assertNull( buffer ); buffer = serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 0, -1, 0 ); assertNull( buffer ); buffer = serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, -1, -1, -1 ); assertNull( buffer ); }
@Test public void testIgnoredExceptionOnLog() throws Exception { Object[] data = new Object[] { 0 }; doThrow( new KettleValueException() ).when( rowMetaInterface ).getString( data ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 10 ); testBufferAux( rowMetaInterface, data, false ); }
@Test public void testGetBufferWindowRowSize() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 10 ); testBufferAux(); }
@Test public void testGetBufferWindowRowSizeRate() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); testBufferAux(); }
@Test public void testGetBufferWindowSizeTime() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 0, 10000 ); testBufferAux(); }
@Test public void testGetBufferWindowSizeTimeRate() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 1000, 10000 ); testBufferAux(); }
@Test public void testGetBufferTime2Calls() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 0, 10000 ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 0, 10000 ); testBufferAux(); }
@Test public void testGetBufferRows2Calls() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 10 ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 0, 10 ); testBufferAux(); }
@Test public void testGetBufferTimeRate2Calls() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 1000, 10000 ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_TIME_BASED, 1000, 1000, 10000 ); testBufferAux(); }
@Test public void testGetBufferRowsRate2Calls() throws Exception { serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); testBufferAux(); }
@Test public void testGetDistinctBuffers() throws Exception { RowMetaInterface rowMeta = serviceTrans.getTransMeta().getStepFields( MOCK_SERVICE_STEP_NAME ); Object[] data = new Object[] { 0 }; serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); testBufferAux( rowMeta, data, false ); serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 2 ); testBufferAux( rowMeta, data, false ); }
@Test public void testStartServiceDataServiceExecutorNull() throws Exception { doReturn( null ).when( context ).getExecutor( any() ); RowMetaInterface rowMeta = serviceTrans.getTransMeta().getStepFields( MOCK_SERVICE_STEP_NAME ); Object[] data = new Object[] { 0 }; serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); testBufferAux( rowMeta, data, false ); }
@Test public void testStartServiceDataServiceExecutor() throws Exception { DataServiceExecutor dataServiceExecutor = mock( DataServiceExecutor.class ); doReturn( dataServiceExecutor ).when( context ).getExecutor( any() ); RowMetaInterface rowMeta = serviceTrans.getTransMeta().getStepFields( MOCK_SERVICE_STEP_NAME ); Object[] data = new Object[] { 0 }; serviceExecutor.getBuffer( MOCK_QUERY, windowConsumer, MOCK_WINDOW_MODE_ROW_BASED, 1, 1, 10 ); testBufferAux( rowMeta, data, false ); verify( dataServiceExecutor ).getParameters(); }