@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;
    }
@ Override public Configuration visit ( AstConnectNode connectNode , State state ) { state . readUnmasker = Masker . IDENTITY_MASKER ; state . writeMasker = Masker . IDENTITY_MASKER ; state . pipelineAsMap = new LinkedHashMap < > ( ) ; for ( AstStreamableNode streamable : connectNode . getStreamables ( ) ) { streamable . accept ( this , state ) ; } 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 ) ; ChannelPipelineFactory pipelineFactory = new ChannelPipelineFactory ( ) { private int numCalled ; @ Override public ChannelPipeline getPipeline ( ) { if ( numCalled ++ != 0 ) { throw new RobotException ( "getPipeline called more than once" ) ; } return pipeline ; } } ; Supplier < URI > locationResolver = connectNode . getLocation ( ) :: getValue ; OptionsResolver optionsResolver = new OptionsResolver ( connectNode . getOptions ( ) ) ; ClientBootstrapResolver clientResolver = new ClientBootstrapResolver ( bootstrapFactory , addressFactory , pipelineFactory , locationResolver , optionsResolver , awaitBarrier , connectNode . getRegionInfo ( ) ) ; 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,