@ Override public final void initialize ( FlowletContext ctx ) throws Exception { super . initialize ( ctx ) ; portsAnnouncementList = Lists . newArrayList ( ) ; DefaultInputFlowletConfigurer configurer = new DefaultInputFlowletConfigurer ( this ) ; create ( configurer ) ; InputFlowletSpecification spec = configurer . createInputFlowletSpec ( ) ; dataIngestionPortsMap = Maps . newHashMap ( ) ; int httpPort = 0 ; if ( ctx . getRuntimeArguments ( ) . get ( Constants . HTTP_PORT ) != null ) { httpPort = Integer . parseInt ( ctx . getRuntimeArguments ( ) . get ( Constants . HTTP_PORT ) ) ; } dataIngestionPortsMap . put ( Constants . HTTP_PORT , httpPort ) ; for ( String inputName : spec . getInputSchemas ( ) . keySet ( ) ) { int tcpPort = 0 ; if ( ctx . getRuntimeArguments ( ) . get ( Constants . TCP_INGESTION_PORT_PREFIX + inputName ) != null ) { tcpPort = Integer . parseInt ( ctx . getRuntimeArguments ( ) . get ( Constants . TCP_INGESTION_PORT_PREFIX + inputName ) ) ; } dataIngestionPortsMap . put ( Constants . TCP_INGESTION_PORT_PREFIX + inputName , tcpPort ) ; } // Setup temporary directory structure tmpFolder = Files . createTempDir ( ) ; File baseDir = new File ( tmpFolder , "baseDir" ) ; baseDir . mkdirs ( ) ; InputFlowletConfiguration inputFlowletConfiguration = new LocalInputFlowletConfiguration ( baseDir , spec ) ; File binDir = inputFlowletConfiguration . createStreamEngineProcesses ( ) ; healthInspector = new HealthInspector ( this ) ; metricsRecorder = new MetricsRecorder ( metrics ) ; //Initiating AbstractInputFlowlet Components recordQueue = new GDATRecordQueue ( ) ; //Initiating Netty TCP I/O ports inputFlowletService = new InputFlowletService ( binDir , spec , healthInspector , metricsRecorder , recordQueue , dataIngestionPortsMap , this ) ; inputFlowletService . startAndWait ( ) ; //Starting health monitor service healthInspector . startAndWait ( ) ; //Initializing methodsDriver Map < String , StreamSchema > schemaMap = MetaInformationParser . getSchemaMap ( new File ( binDir . toURI ( ) ) ) ; methodsDriver = new MethodsDriver ( this , schemaMap ) ; //Initialize stopwatch and retry counter stopwatch = new Stopwatch ( ) ; retryCounter = 0 ; }
This method initializes all the components required to setup the SQL Compiler environment .