DataServiceClient implements IDataServiceClientService { @Override public List<IThinServiceInformation> getServiceInformation() throws SQLException { List<IThinServiceInformation> services = Lists.newArrayList(); for ( DataServiceMeta service : resolver.getDataServices( logErrors()::apply ) ) { TransMeta transMeta = service.getServiceTrans(); try { transMeta.activateParameters(); RowMetaInterface serviceFields = transMeta.getStepFields( service.getStepname() ); IThinServiceInformation serviceInformation = new ThinServiceInformation( service.getName(), service.isStreaming(), serviceFields ); services.add( serviceInformation ); } catch ( Exception e ) { String message = MessageFormat.format( "Unable to get fields for service {0}, transformation: {1}", service.getName(), transMeta.getName() ); log.logError( message, e ); } } return services; } DataServiceClient( Query.Service queryService, DataServiceResolver resolver,
                            ExecutorService executorService ); @Override DataInputStream query( String sqlQuery, final int maxRows ); @Override DataInputStream query( String sqlQuery, final int maxRows,
                                          Map<String, String> params ); @Override DataInputStream query( String sqlQuery, IDataServiceClientService.StreamingMode windowMode,
                                          long windowSize, long windowEvery,
                                          long windowLimit ); @Override DataInputStream query( String sqlQuery, IDataServiceClientService.StreamingMode windowMode,
                                          long windowSize, long windowEvery,
                                          long windowLimit, Map<String, String> params ); Query prepareQuery( String sql, int maxRows, Map<String, String> parameters ); Query prepareQuery( String sql, IDataServiceClientService.StreamingMode windowMode,
                             long windowSize, long windowEvery, long windowLimit, Map<String, String> parameters ); @Override List<IThinServiceInformation> getServiceInformation(); @Override ThinServiceInformation getServiceInformation( String name ); @Override List<String> getServiceNames( String serviceName ); @Override List<String> getServiceNames(); List<DataServiceMeta> getServiceMeta(); DataServiceMeta getServiceMeta( String serviceName ); void setLogChannel( LogChannelInterface log ); LogChannelInterface getLogChannel(); @Deprecated void setRepository( Repository repository ); @Deprecated void setMetaStore( IMetaStore metaStore ); @Override void query( String sqlQuery, IStreamingParams streamParams,
                     Map<String, String> params, Observer<List<RowMetaAndData>> consumer );  }
@Test public void testGetServiceInformation() throws Exception { when( transMeta.getStepFields( dataService.getStepname() ) ).thenReturn( rowMetaInterface ); when( dataServiceResolver.getDataServices( anyString(), any() ) ).thenReturn( ImmutableList.of( dataService ) ); when( dataServiceResolver.getDataServices( any() ) ).thenReturn( ImmutableList.of( dataService ) ); assertThat( client.getServiceInformation(), contains( allOf( hasProperty( "name", equalTo( DATA_SERVICE_NAME ) ), hasProperty( "serviceFields", equalTo( rowMetaInterface ) ), hasProperty( "streaming", equalTo( false ) ) ) ) ); verify( transMeta ).activateParameters(); when( transMeta.getStepFields( DATA_SERVICE_STEP ) ).thenThrow( new KettleStepException() ); assertThat( client.getServiceInformation(), is( empty() ) ); }
@Test public void testWindowGetServiceInformation() throws Exception { when( streamingTransMeta.getStepFields( streamingDataService.getStepname() ) ).thenReturn( rowMetaInterface ); when( dataServiceResolver.getDataServices( anyString(), any() ) ).thenReturn( ImmutableList.of( streamingDataService ) ); when( dataServiceResolver.getDataServices( any() ) ).thenReturn( ImmutableList.of( streamingDataService ) ); assertThat( client.getServiceInformation(), contains( allOf( hasProperty( "name", equalTo( STREAMING_DATA_SERVICE_NAME ) ), hasProperty( "serviceFields", equalTo( rowMetaInterface ) ), hasProperty( "streaming", equalTo( true ) ) ) ) ); verify( streamingTransMeta ).activateParameters(); when( streamingTransMeta.getStepFields( DATA_SERVICE_STEP ) ).thenThrow( new KettleStepException() ); assertThat( client.getServiceInformation(), is( empty() ) ); }
@Test public void testGetServiceInformationByName() throws Exception { when( transMeta.getStepFields( dataService.getStepname() ) ).thenReturn( rowMetaInterface ); ThinServiceInformation serviceInformation = client.getServiceInformation( DATA_SERVICE_NAME ); verify( transMeta ).activateParameters(); assertThat( serviceInformation.getName(), equalTo( DATA_SERVICE_NAME ) ); assertThat( serviceInformation.getServiceFields(), equalTo( rowMetaInterface ) ); }