import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public <T extends DataFlowNode> List<String> getPropertyNames(Class<T> dataFlowNodeClass, Map<String, String> metaProperties)
    throws InvalidClassException, InvalidMetaPropertyException, MissingMetaPropertyException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
        {
            List<String> propertyNames = new LinkedList<String>();

            return propertyNames;
        }
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Test
public void simplestConversion()
{
    DataFlowNodeLifeCycleControl dataFlowNodeLifeCycleControl = new TestJEEDataFlowNodeLifeCycleControl();

    DummyDataSource dummyDataSource        = new DummyDataSource("Dummy Data Source", Collections.<String, String>emptyMap());
    DataProcessor   shapeFileDataProcessor = new ShapeFileConverterDataProcessor("ShapeFile Converter Data Processor", Collections.<String, String>emptyMap());
    DummyDataSink   dummyDataSink          = new DummyDataSink("Dummy Data Sink", Collections.<String, String>emptyMap());

    dataFlowNodeLifeCycleControl.completeCreationAndActivateDataFlowNode(UUID.randomUUID().toString(), dummyDataSource, null);
    dataFlowNodeLifeCycleControl.completeCreationAndActivateDataFlowNode(UUID.randomUUID().toString(), shapeFileDataProcessor, null);
    dataFlowNodeLifeCycleControl.completeCreationAndActivateDataFlowNode(UUID.randomUUID().toString(), dummyDataSink, null);

    ((ObservableDataProvider<File>) dummyDataSource.getDataProvider(File.class)).addDataConsumer((ObserverDataConsumer<File>) shapeFileDataProcessor.getDataConsumer(File.class));
    ((ObservableDataProvider<String>) shapeFileDataProcessor.getDataProvider(String.class)).addDataConsumer((ObserverDataConsumer<String>) dummyDataSink.getDataConsumer(String.class));

    File testFile = new File("/tmp/Gully_point/Gully_point.shp");
    dummyDataSource.sendData(testFile);

    dataFlowNodeLifeCycleControl.removeDataFlowNode(dummyDataSource);
    dataFlowNodeLifeCycleControl.removeDataFlowNode(shapeFileDataProcessor);
    dataFlowNodeLifeCycleControl.removeDataFlowNode(dummyDataSink);

}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public List<Class<? extends DataFlowNode>> getClasses()
{
    List<Class<? extends DataFlowNode>> classes = new LinkedList<Class<? extends DataFlowNode>>();

    classes.add(DataProcessor.class);

    return classes;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public <T extends DataFlowNode> List<String> getMetaPropertyNames(Class<T> dataFlowNodeClass)
    throws InvalidClassException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        List<String> metaPropertyNames = new LinkedList<String>();

        return metaPropertyNames;
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public <T extends DataFlowNode> T createDataFlowNode(String name, Class<T> dataFlowNodeClass, Map<String, String> metaProperties, Map<String, String> properties)
    throws InvalidClassException, InvalidNameException, InvalidMetaPropertyException, MissingMetaPropertyException, InvalidPropertyException, MissingPropertyException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
            return (T) new XSSFStreamSheetToCSVDataProcessor(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public <T extends DataFlowNode> T createDataFlowNode(String name, Class<T> dataFlowNodeClass, Map<String, String> metaProperties, Map<String, String> properties)
    throws InvalidClassException, InvalidNameException, InvalidMetaPropertyException, MissingMetaPropertyException, InvalidPropertyException, MissingPropertyException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
            return (T) new XSSFSheetToCSVDataProcessor(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public <T extends DataFlowNode> T createDataFlowNode(String name, Class<T> dataFlowNodeClass, Map<String, String> metaProperties, Map<String, String> properties)
    throws InvalidClassException, InvalidNameException, InvalidMetaPropertyException, MissingMetaPropertyException, InvalidPropertyException, MissingPropertyException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
            return (T) new XSSFRowToJSONDataProcessor(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public <T extends DataFlowNode> T createDataFlowNode(String name, Class<T> dataFlowNodeClass, Map<String, String> metaProperties, Map<String, String> properties)
    throws InvalidClassException, InvalidNameException, InvalidMetaPropertyException, MissingMetaPropertyException, InvalidPropertyException, MissingPropertyException
{
    if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
            return (T) new FileReaderDataProcessor(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public List<Class<? extends DataFlowNode>> getClasses()
{
    List<Class<? extends DataFlowNode>> classes = new LinkedList<Class<? extends DataFlowNode>>();

    classes.add(DataSource.class);
    classes.add(DataProcessor.class);
    classes.add(DataSink.class);

    return classes;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public <T extends DataFlowNode> List<String> getMetaPropertyNames(Class<T> dataFlowNodeClass)
    throws InvalidClassException
{
    if (dataFlowNodeClass.equals(DataSource.class) || dataFlowNodeClass.equals(DataProcessor.class) || dataFlowNodeClass.equals(DataSink.class))
        return Collections.emptyList();
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public <T extends DataFlowNode> T createDataFlowNode(String name, Class<T> dataFlowNodeClass, Map<String, String> metaProperties, Map<String, String> properties)
    throws InvalidClassException, InvalidNameException, InvalidMetaPropertyException, MissingMetaPropertyException, InvalidPropertyException, MissingPropertyException
{
    if (dataFlowNodeClass.equals(DataSource.class))
    {
        if (metaProperties.isEmpty())
            return (T) new PullSOAPWebServiceDataSource(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else if (dataFlowNodeClass.equals(DataProcessor.class))
    {
        if (metaProperties.isEmpty())
            return (T) new Document2TextDataProcessor(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else if (dataFlowNodeClass.equals(DataSink.class))
    {
        if (metaProperties.isEmpty())
            return (T) new PushSOAPWebServiceDataSink(name, properties);
        else
            throw new InvalidMetaPropertyException("No metaproperties expected", null, null);
    }
    else
        throw new InvalidClassException("Unsupported class", dataFlowNodeClass.getName());
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@GET
@Produces(MediaType.APPLICATION_JSON)
public DataBrokerDTO getDataBrokerSummariesJSON()
{
    logger.log(Level.FINE, "DataBrokerWS.getDataBrokerSummariesJSON");

    if ((_dataFlowInventory != null) && (_dataFlowNodeFactoryInventory != null))
    {
        DataBrokerDTO dataBrokerDTO = new DataBrokerDTO();

        List<String> dataFlowNames = new LinkedList<String>();
        for (DataFlow dataFlow: _dataFlowInventory.getDataFlows())
            dataFlowNames.add(dataFlow.getName());
        dataBrokerDTO.setDataFlowNames(dataFlowNames);

        List<DataFlowNodeFactoryDTO> dataFlowNodeFactories = new LinkedList<DataFlowNodeFactoryDTO>();
        for (DataFlowNodeFactory dataFlowNodeFactory: _dataFlowNodeFactoryInventory.getDataFlowNodeFactorys())
        {
            List<Class<? extends DataFlowNode>> dataFlowNodeFactoryClasses = dataFlowNodeFactory.getClasses();

            Boolean dataSourceFactory    = dataFlowNodeFactoryClasses.contains(DataSource.class);
            Boolean dataSinkFactory      = dataFlowNodeFactoryClasses.contains(DataSink.class);;
            Boolean dataProcessorFactory = dataFlowNodeFactoryClasses.contains(DataProcessor.class);;
            Boolean dataServiceFactory   = dataFlowNodeFactoryClasses.contains(DataService.class);;
            Boolean dataStoreFactory     = dataFlowNodeFactoryClasses.contains(DataStore.class);;

            dataFlowNodeFactories.add(new DataFlowNodeFactoryDTO(dataFlowNodeFactory.getName(), dataFlowNodeFactory.getProperties(), dataSourceFactory, dataSinkFactory, dataProcessorFactory, dataServiceFactory, dataStoreFactory));
        }
        dataBrokerDTO.setDataFlowNodeFactories(dataFlowNodeFactories);

        return dataBrokerDTO;
    }
    else
        throw new WebApplicationException(HttpURLConnection.HTTP_INTERNAL_ERROR);
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
public static String classToString(Class<? extends DataFlowNode> dataFlowNodeClass)
{
    if (DataSource.class.isAssignableFrom(dataFlowNodeClass))
        return DATASOURCE_TYPE;
    else if (DataSink.class.isAssignableFrom(dataFlowNodeClass))
        return DATASINK_TYPE;
    else if (DataService.class.isAssignableFrom(dataFlowNodeClass))
        return DATASERVICE_TYPE;
    else if (DataStore.class.isAssignableFrom(dataFlowNodeClass))
        return DATASTORE_TYPE;
    else if (DataProcessor.class.isAssignableFrom(dataFlowNodeClass))
        return DATAPROCESSOR_TYPE;
    else
        return null;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
public static Class<? extends DataFlowNode> stringToClass(String dataFlowNodeClassName)
{
    if (DATASOURCE_TYPE.equals(dataFlowNodeClassName))
        return DataSource.class;
    else if (DATASINK_TYPE.equals(dataFlowNodeClassName))
        return DataSink.class;
    else if (DATASERVICE_TYPE.equals(dataFlowNodeClassName))
        return DataService.class;
    else if (DATASTORE_TYPE.equals(dataFlowNodeClassName))
        return DataStore.class;
    else if (DATAPROCESSOR_TYPE.equals(dataFlowNodeClassName))
        return DataProcessor.class;
    else
        return null;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
private Collection<Class<?>> getSourceProviderClasses(DataFlowNode sourceDataFlowNode)
{
    if (sourceDataFlowNode instanceof DataSource)
        return ((DataSource) sourceDataFlowNode).getDataProviderDataClasses();
    else if (sourceDataFlowNode instanceof DataProcessor)
        return ((DataProcessor) sourceDataFlowNode).getDataProviderDataClasses();
    else if (sourceDataFlowNode instanceof DataService)
        return ((DataService) sourceDataFlowNode).getDataProviderDataClasses();
    else if (sourceDataFlowNode instanceof DataStore)
        return ((DataStore) sourceDataFlowNode).getDataProviderDataClasses();
    else
        return Collections.emptySet();
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
private Collection<Class<?>> getSinkConsumerClasses(DataFlowNode sinkDataFlowNode)
{
    if (sinkDataFlowNode instanceof DataProcessor)
        return ((DataProcessor) sinkDataFlowNode).getDataConsumerDataClasses();
    else if (sinkDataFlowNode instanceof DataService)
        return ((DataService) sinkDataFlowNode).getDataConsumerDataClasses();
    else if (sinkDataFlowNode instanceof DataStore)
        return ((DataStore) sinkDataFlowNode).getDataConsumerDataClasses();
    else if (sinkDataFlowNode instanceof DataSink)
        return ((DataSink) sinkDataFlowNode).getDataConsumerDataClasses();
    else
        return Collections.emptySet();
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
private <T> DataProvider<T> getSourceProvider(DataFlowNode sourceDataFlowNode, Class<T> dataClass)
{
    if (sourceDataFlowNode instanceof DataSource)
        return ((DataSource) sourceDataFlowNode).getDataProvider(dataClass);
    else if (sourceDataFlowNode instanceof DataProcessor)
        return ((DataProcessor) sourceDataFlowNode).getDataProvider(dataClass);
    else if (sourceDataFlowNode instanceof DataService)
        return ((DataService) sourceDataFlowNode).getDataProvider(dataClass);
    else if (sourceDataFlowNode instanceof DataStore)
        return ((DataStore) sourceDataFlowNode).getDataProvider(dataClass);
    else
        return null;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
private <T> DataConsumer<T> getSinkConsumer(DataFlowNode sinkDataFlowNode, Class<T> dataClass)
{
    if (sinkDataFlowNode instanceof DataProcessor)
        return ((DataProcessor) sinkDataFlowNode).getDataConsumer(dataClass);
    else if (sinkDataFlowNode instanceof DataService)
        return ((DataService) sinkDataFlowNode).getDataConsumer(dataClass);
    else if (sinkDataFlowNode instanceof DataStore)
        return ((DataStore) sinkDataFlowNode).getDataConsumer(dataClass);
    else if (sinkDataFlowNode instanceof DataSink)
        return ((DataSink) sinkDataFlowNode).getDataConsumer(dataClass);
    else
        return null;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@Override
public List<Class<? extends DataFlowNode>> getClasses()
{
    List<Class<? extends DataFlowNode>> classes = new LinkedList<Class<? extends DataFlowNode>>();
    
    classes.add(DataSource.class);
    classes.add(DataProcessor.class);
    classes.add(DataSink.class);
    classes.add(DataService.class);
    classes.add(DataStore.class);
    
    return classes;
}
 

import com.arjuna.databroker.data.DataProcessor; //导入依赖的package包/类
@GET
@Path("{dataflowid}/_factoryinfo")
@Produces(MediaType.APPLICATION_JSON)
public DataFlowNodeFactoryDTO getFactoryInfoJSON(@PathParam("dataflowid") String dataFlowId, @QueryParam("factoryname") String factoryName)
{
    logger.log(Level.FINE, "DataFlowWS.getFactoryInfoJSON");

    if (_dataFlowInventory != null)
    {
        if (dataFlowId != null)
        {
            DataFlow dataFlow = _dataFlowInventory.getDataFlow(dataFlowId);

            if ((dataFlow != null) && (dataFlow.getDataFlowNodeFactoryInventory() !=  null))
            {
                DataFlowNodeFactory dataFlowNodeFactory = dataFlow.getDataFlowNodeFactoryInventory().getDataFlowNodeFactory(factoryName);

                if (dataFlowNodeFactory != null)
                {
                    try
                    {
                        List<Class<? extends DataFlowNode>> dataFlowNodeFactoryClasses = dataFlowNodeFactory.getClasses();

                        Boolean dataSourceFactory    = dataFlowNodeFactoryClasses.contains(DataSource.class);
                        Boolean dataSinkFactory      = dataFlowNodeFactoryClasses.contains(DataSink.class);;
                        Boolean dataProcessorFactory = dataFlowNodeFactoryClasses.contains(DataProcessor.class);;
                        Boolean dataServiceFactory   = dataFlowNodeFactoryClasses.contains(DataService.class);;
                        Boolean dataStoreFactory     = dataFlowNodeFactoryClasses.contains(DataStore.class);;

                        return new DataFlowNodeFactoryDTO(dataFlowNodeFactory.getName(), dataFlowNodeFactory.getProperties(), dataSourceFactory, dataSinkFactory, dataProcessorFactory, dataServiceFactory, dataStoreFactory);
                    }
                    catch (Throwable throwable)
                    {
                        logger.log(Level.WARNING, "Problem while obtaining meta property names from data flow node factory: ", throwable);

                        throw new WebApplicationException(HttpURLConnection.HTTP_INTERNAL_ERROR);
                    }
                }
                else
                    throw new WebApplicationException(HttpURLConnection.HTTP_NOT_FOUND);
            }
            else
                throw new WebApplicationException(HttpURLConnection.HTTP_NOT_FOUND);
        }
        else
            throw new WebApplicationException(HttpURLConnection.HTTP_NOT_FOUND);
    }
    else
        throw new WebApplicationException(HttpURLConnection.HTTP_INTERNAL_ERROR);
}