DataServiceExecutor { public void stop( boolean stopTrans ) { synchronized ( getServiceTrans() ) { if ( stopTrans || !service.isStreaming() ) { if ( serviceTrans.isRunning() ) { serviceTrans.stopAll(); } if ( genTrans.isRunning() ) { genTrans.stopAll(); } if ( service.isStreaming() ) { this.genTransformationPushBasedIsFinished.set( true ); StreamingServiceTransExecutor serviceExecutor = context.getServiceTransExecutor( streamServiceKey ); if ( serviceExecutor != null ) { String streamingGenTransCacheKey = getStreamingGenTransCacheKey(); context.removeStreamingGeneratedTransExecution( streamingGenTransCacheKey ); serviceExecutor.clearCacheByKey( streamingGenTransCacheKey ); } } } } } private  DataServiceExecutor( Builder builder ); static void writeMetadata( DataOutputStream dos, String... metadatas ); DataServiceExecutor executeQuery( final DataOutputStream dos ); DataServiceExecutor executeQuery( Observer<RowMetaAndData> consumer ); DataServiceExecutor executeStreamingQuery( final Observer<List<RowMetaAndData>> streamingConsumer, boolean pollingMode ); String getStreamingGenTransCacheKey(); DataServiceExecutor executeDefaultQuery( final Observer<RowMetaAndData> consumer ); DataServiceExecutor executeQuery(); void executeListeners( ExecutionPoint... stages ); RowProducer addRowProducer(); void waitUntilFinished(); TransMeta getServiceTransMeta(); TransMeta getGenTransMeta(); DataServiceMeta getService(); Trans getServiceTrans(); Trans getGenTrans(); String getServiceName(); Map<String, String> getParameters(); String getId(); Boolean hasErrors(); static String calculateTransname( SQL sql, boolean isService ); void stop( boolean stopTrans ); boolean isStopped(); SQL getSql(); String getResultStepName(); int getRowLimit(); int getServiceRowLimit(); ListMultimap<ExecutionPoint, Runnable> getListenerMap();  }
@Test public void testStop() throws KettleException { String sql = "SELECT * FROM " + DATA_SERVICE_NAME; when( serviceTrans.isRunning() ).thenReturn( true ); when( genTrans.isRunning() ).thenReturn( true ); DataServiceExecutor executor = new DataServiceExecutor.Builder( new SQL( sql ), dataService, context ). serviceTrans( serviceTrans ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). build(); executor.stop( true ); verify( serviceTrans ).stopAll(); verify( genTrans ).stopAll(); }
@Test public void testStopMixedconditions() throws KettleException { String sql = "SELECT * FROM " + DATA_SERVICE_NAME; when( serviceTrans.isRunning() ).thenReturn( true ); when( genTrans.isRunning() ).thenReturn( true ); DataServiceExecutor executor = new DataServiceExecutor.Builder( new SQL( sql ), dataService, context ). serviceTrans( serviceTrans ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). build(); when( serviceTrans.isRunning() ).thenReturn( false ); when( genTrans.isRunning() ).thenReturn( true ); executor.stop( true ); verify( serviceTrans, times( 0 ) ).stopAll(); verify( genTrans, times( 1 ) ).stopAll(); when( serviceTrans.isRunning() ).thenReturn( true ); when( genTrans.isRunning() ).thenReturn( false ); executor.stop( true ); verify( serviceTrans, times( 1 ) ).stopAll(); verify( genTrans, times( 1 ) ).stopAll(); }
@Test public void testStopWhenAlreadyStopped() throws KettleException { String sql = "SELECT * FROM " + DATA_SERVICE_NAME; when( serviceTrans.isRunning() ).thenReturn( false ); when( genTrans.isRunning() ).thenReturn( false ); DataServiceExecutor executor = new DataServiceExecutor.Builder( new SQL( sql ), dataService, context ). serviceTrans( serviceTrans ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). build(); executor.stop( true ); verify( serviceTrans, times( 0 ) ).stopAll(); verify( genTrans, times( 0 ) ).stopAll(); }