public static StormTopology buildTopology ( ExecutionContext context ) throws IllegalAccessException , InstantiationException , ClassNotFoundException , NoSuchMethodException , InvocationTargetException { StormTopology topology = null ; TopologyDef topologyDef = context . getTopologyDef ( ) ; if ( ! topologyDef . validate ( ) ) { throw new IllegalArgumentException ( "Invalid topology config. Spouts, bolts and streams cannot be " + "defined in the same configuration as a topologySource." ) ; } // build components that may be referenced by spouts, bolts, etc. // the map will be a String --> Object where the object is a fully // constructed class instance buildComponents ( context ) ; if ( topologyDef . isDslTopology ( ) ) { // This is a DSL (YAML, etc.) topology... LOG . info ( "Detected DSL topology..." ) ; TopologyBuilder builder = new TopologyBuilder ( ) ; // create spouts buildSpouts ( context , builder ) ; // we need to be able to lookup bolts by id, then switch based // on whether they are IBasicBolt or IRichBolt instances buildBolts ( context ) ; // process stream definitions buildStreamDefinitions ( context , builder ) ; topology = builder . createTopology ( ) ; } else { // user class supplied... // this also provides a bridge to Trident... LOG . info ( "A topology source has been specified..." ) ; ObjectDef def = topologyDef . getTopologySource ( ) ; topology = buildExternalTopology ( def , context ) ; } return topology ; }
Given a topology definition return a Storm topology that can be run either locally or remotely .