@ Override public Configuration visit ( AstConnectNode connectNode , State state ) { // masking is a no-op by default for each stream state . readUnmasker = Masker . IDENTITY_MASKER ; state . writeMasker = Masker . IDENTITY_MASKER ; state . pipelineAsMap = new LinkedHashMap <> ( ) ; for ( AstStreamableNode streamable : connectNode . getStreamables ( ) ) { streamable . accept ( this , state ) ; } /* Add the completion handler */ String handlerName = String . format ( "completion#%d" , state . pipelineAsMap . size ( ) + 1 ) ; CompletionHandler completionHandler = new CompletionHandler ( ) ; completionHandler . setRegionInfo ( connectNode . getRegionInfo ( ) ) ; state . pipelineAsMap . put ( handlerName , completionHandler ) ; String awaitName = connectNode . getAwaitName ( ) ; Barrier awaitBarrier = null ; if ( awaitName != null ) { awaitBarrier = state . lookupBarrier ( awaitName ) ; } final ChannelPipeline pipeline = pipelineFromMap ( state . pipelineAsMap ) ; /* * TODO. This is weird. I will only have one pipeline per connect. But if I don't set a factory When a connect * occurs it will create a shallow copy of the pipeline I set. This doesn't work due to the beforeAdd methods in * ExecutionHandler. Namely when the pipeline is cloned it uses the same handler objects so the handler future * is not null and we fail with an assertion error. */ ChannelPipelineFactory pipelineFactory = new ChannelPipelineFactory ( ) { private int numCalled ; @ Override public ChannelPipeline getPipeline ( ) { if ( numCalled ++ != 0 ) { throw new RobotException ( "getPipeline called more than once" ) ; } return pipeline ; } } ; // Now that connect supports barrier and expression value, connect uri may not be available at this point. // To defer the evaluation of connect uri and initialization of ClientBootstrap, LocationResolver and // ClientResolver are created with information necessary to create ClientBootstrap when the connect uri // is available. Supplier < URI > locationResolver = connectNode . getLocation ( ) :: getValue ; OptionsResolver optionsResolver = new OptionsResolver ( connectNode . getOptions ( ) ) ; ClientBootstrapResolver clientResolver = new ClientBootstrapResolver ( bootstrapFactory , addressFactory , pipelineFactory , locationResolver , optionsResolver , awaitBarrier , connectNode . getRegionInfo ( ) ) ; // retain pipelines for tear down state . configuration . getClientAndServerPipelines ( ) . add ( pipeline ) ; state . configuration . getClientResolvers ( ) . add ( clientResolver ) ; return state . configuration ; }
Creates the pipeline visits all streamable nodes and the creates the ClientBootstrap with the pipeline and remote address