/*
@see com.ibm.wsspi.channelfw.ChannelFactory#findOrCreateChannel(ChannelData)
331822
@SuppressWarnings("unused")
    public synchronized Channel findOrCreateChannel(ChannelData channelData) throws ChannelException {
        String channelName = channelData.getName();
        Channel ret = this.existingChannels.get(channelName);
        if (ret == null) {
            // Create the new channel with the input configuration
            if (null != this.globalBufferSize) {
                // if the factory setting exists, see if the channel overrides it
                // otherwise apply it now
                Map<Object, Object> props = channelData.getPropertyBag();
                if (!props.containsKey(PROP_BUFFERSIZE)) {
                    props.put(PROP_BUFFERSIZE, this.globalBufferSize);
                }
            }
            ret = new HttpDispatcherChannel(channelData, this);
            this.existingChannels.put(channelName, ret);
        }
        return ret;
    }
30754
private void buildChannel() {
      FileLookup fileLookup = FileLookupFactory.newInstance();

      // in order of preference - we first look for an external JGroups file, then a set of XML
      // properties, and
      // finally the legacy JGroups String properties.
      String cfg;
      if (props != null) {
         if (props.containsKey(CHANNEL_LOOKUP)) {
            String channelLookupClassName = props.getProperty(CHANNEL_LOOKUP);

            try {
               JGroupsChannelLookup lookup = Util.getInstance(channelLookupClassName, configuration.classLoader());
               channel = lookup.getJGroupsChannel(props);
               connectChannel = lookup.shouldConnect();
               disconnectChannel = lookup.shouldDisconnect();
               closeChannel = lookup.shouldClose();
            } catch (ClassCastException e) {
               log.wrongTypeForJGroupsChannelLookup(channelLookupClassName, e);
               throw new CacheException(e);
            } catch (Exception e) {
               log.errorInstantiatingJGroupsChannelLookup(channelLookupClassName, e);
               throw new CacheException(e);
            }
         }

         if (channel == null && props.containsKey(CHANNEL_CONFIGURATOR)) {
            JGroupsChannelConfigurator configurator = (JGroupsChannelConfigurator) props.get(CHANNEL_CONFIGURATOR);
            try {
               channel = configurator.createChannel();
            } catch (Exception e) {
               throw log.errorCreatingChannelFromConfigurator(configurator.getProtocolStackString(), e);
            }
         }

         if (channel == null && props.containsKey(CONFIGURATION_FILE)) {
            cfg = props.getProperty(CONFIGURATION_FILE);
            Collection<URL> confs = Collections.emptyList();
            try {
               confs = fileLookup.lookupFileLocations(cfg, configuration.classLoader());
            } catch (IOException io) {
               //ignore, we check confs later for various states
            }
            if (confs.isEmpty()) {
               throw log.jgroupsConfigurationNotFound(cfg);
            } else if (confs.size() > 1) {
               log.ambiguousConfigurationFiles(Util.toStr(confs));
            }
            try {
               channel = new JChannel(confs.iterator().next());
            } catch (Exception e) {
               throw log.errorCreatingChannelFromConfigFile(cfg, e);
            }
         }

         if (channel == null && props.containsKey(CONFIGURATION_XML)) {
            cfg = props.getProperty(CONFIGURATION_XML);
            try {
               channel = new JChannel(XmlConfigHelper.stringToElement(cfg));
            } catch (Exception e) {
               throw log.errorCreatingChannelFromXML(cfg, e);
            }
         }

         if (channel == null && props.containsKey(CONFIGURATION_STRING)) {
            cfg = props.getProperty(CONFIGURATION_STRING);
            try {
               channel = new JChannel(new ByteArrayInputStream(cfg.getBytes()));
            } catch (Exception e) {
               throw log.errorCreatingChannelFromConfigString(cfg, e);
            }
         }
      }

      if (channel == null) {
         log.unableToUseJGroupsPropertiesProvided(props);
         try {
            channel = new JChannel(
                  fileLookup.lookupFileLocation(DEFAULT_JGROUPS_CONFIGURATION_FILE, configuration.classLoader()));
         } catch (Exception e) {
            throw log.errorCreatingChannelFromConfigFile(DEFAULT_JGROUPS_CONFIGURATION_FILE, e);
         }
      }
   }
1083969
public function channel($channel_id, $source_factory = null, $emitter_factory = null)
    {
        if (isset($this->channel_list[$channel_id])){
            return $this->channel_list[$channel_id];
        }

        $source = null;
        if ($source_factory){
            if (is_callable($source_factory)){
                $source = call_user_func($source_factory);
            }
            else if ($source_factory instanceof EventSourceFactoryInterface){
                $source = $source_factory->createEventSource();
            }
        }

        $emitter = null;
        if ($emitter_factory){
            if (is_callable($emitter_factory)){
                $emitter = call_user_func($emitter_factory);
            }
            else if ($emitter_factory instanceof EventSourceFactoryInterface){
                $emitter = $emitter_factory->createEventEmitter();
            }
        }

        $channel = new EventChannel($source, $emitter);
        $this->channel_list[$channel_id] = $channel;

        return $channel;
    }
46539
public ChannelCategory getOrCreateChannelCategory(JsonNode data) {
        long id = Long.parseLong(data.get("id").asText());
        ChannelType type = ChannelType.fromId(data.get("type").asInt());
        synchronized (this) {
            if (type == ChannelType.CHANNEL_CATEGORY) {
                return getChannelCategoryById(id).orElseGet(() -> new ChannelCategoryImpl(api, this, data));
            }
        }
        // Invalid channel type
        return null;
    }
974052
static void checkChannelValidity(Object channel) throws IOException {
    if (channel == null) {
      /* Most common reason is that original socket does not have a channel.
       * So making this an IOException rather than a RuntimeException.
       */
      throw new IOException("Channel is null. Check " +
                            "how the channel or socket is created.");
    }
    
    if (!(channel instanceof SelectableChannel)) {
      throw new IOException("Channel should be a SelectableChannel");
    }    
  }
814301
public ServerTextChannel getOrCreateServerTextChannel(JsonNode data) {
        long id = Long.parseLong(data.get("id").asText());
        ChannelType type = ChannelType.fromId(data.get("type").asInt());
        synchronized (this) {
            // TODO Treat news channels differently
            if (type == ChannelType.SERVER_TEXT_CHANNEL || type == ChannelType.SERVER_NEWS_CHANNEL) {
                return getTextChannelById(id).orElseGet(() -> new ServerTextChannelImpl(api, this, data));
            }
        }
        // Invalid channel type
        return null;
    }
735310
@Override
    public synchronized ChainData addChain(String chainName, FlowType chainType, String[] channelNameList) throws ChannelException, ChainException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "addChain: " + chainName);
        }
        if (null == chainName) {
            throw new InvalidChainNameException("Input chain name is null");
        }
        if (null == channelNameList || 0 == channelNameList.length) {
            throw new InvalidChannelNameException("Invalid channel list");
        }

        ChainDataImpl chainData = chainDataMap.get(chainName);
        // Ensure the chain name doesn't already exist in the framework.
        if (null != chainData) {
            InvalidChainNameException e = new InvalidChainNameException("Chain config already exists, " + chainName);
            FFDCFilter.processException(e, getClass().getName() + ".addChain", "1411", this, new Object[] { chainData });
            throw e;
        }

        int zRegion = RegionType.NOT_ON_Z;
        boolean okToAdd = true;
        ChannelFactory factory = null;

        // Determine if we are running on Z, and if so, what region we are in
        if (currentlyOnZ()) {
            zRegion = currentZRegion();
        }

        // Ensure each of the channel configurations already exist in the framework.
        ChannelData[] channelDataArray = new ChannelData[channelNameList.length];
        ChannelData channelData = null;
        int i = 0;
        int lastIndex = channelNameList.length - 1;
        for (i = 0; i <= lastIndex; i++) {
            factory = null;
            channelData = channelDataMap.get(channelNameList[i]);
            if (null == channelData) {
                InvalidChannelNameException e = new InvalidChannelNameException("Can't add chain config due to unknown channel, " + channelNameList[i]);
                FFDCFilter.processException(e, getClass().getName() + ".addChain", "1443", this, new Object[] { channelNameList[i] });
                throw e;
            }
            if (zRegion != RegionType.NOT_ON_Z && chainType.equals(FlowType.INBOUND)) {
                // On Z and inbound, need to do some special Z things
                if (i == lastIndex) {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "On Z, looking at Application Channel, " + channelData.getName());
                    }
                    // Looking at the Inbound Application Channel
                    factory = getChannelFactoryInternal(channelData.getFactoryType(), false);

                    if (zRegion == RegionType.SR_REGION) {
                        // in SR, check if we should start this chain
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Operating in Servant Region");
                        }
                        if (factory instanceof BoundRegion) {
                            if (!(((BoundRegion) factory).isServantStartable(channelDataMap))) {
                                // channel does not want to start given the current
                                // configuration
                                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                    Tr.debug(tc, "Channel says it is not startable in Servant Region");
                                }
                                okToAdd = false;
                            }
                        }
                    } else if ((zRegion == BoundRegion.CR_REGION) || (zRegion == BoundRegion.CRA_REGION)) {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Operating in CR or CRA Region, Region constant: " + zRegion);
                        }
                        // In a secure region. check if the channel implements BoundRegion,
                        // and is configured to run in this region
                        if (factory instanceof BoundRegion) {
                            if (zRegion != (((BoundRegion) factory).getRegion(channelDataMap))) {
                                // channel does not support running in this region
                                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                    Tr.debug(tc, "This Channel says it is not startable in this Region: getRegion returned: " + ((BoundRegion) factory).getRegion(channelDataMap));
                                }
                                okToAdd = false;
                            }
                        }
                    }
                } else {
                    // Not Looking at an application channel in the chain
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                        Tr.debug(tc, "On Z, looking at non-application Channel, " + channelData.getName());
                    }

                    if ((zRegion == BoundRegion.CR_REGION) || (zRegion == BoundRegion.CRA_REGION)) {
                        // mark what region this channel is being asked to run in, and
                        // log a warning if this gives a sharable problem
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                            Tr.debug(tc, "Operating in CR or CRA Region, Region constant: " + zRegion);
                        }

                        // Get Channel entry from HashMap
                        Integer channelEntry = ChannelZRegions.get(channelData.getName());
                        if (channelEntry == null) {
                            // new entry
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "First time setting region entry for this channel");
                            }
                            ChannelZRegions.put(channelData.getName(), Integer.valueOf(zRegion));
                        } else {
                            // check to see if we are being asked to run in a different region
                            int entryValue = channelEntry.intValue();

                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                Tr.debug(tc, "Not first time setting region entry for this channel, current region entry is: " + entryValue);
                            }

                            if ((entryValue & zRegion) == 0) {
                                // being asked to be run in a different region, see if this is
                                // sharable
                                factory = getChannelFactoryInternal(channelData.getFactoryType(), false);
                                if (factory instanceof CrossRegionSharable) {
                                    if (!(((CrossRegionSharable) factory).isSharable(channelDataMap))) {
                                        // non-sharable channel is being put into different regions
                                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                                            Tr.debug(tc, "Channel says it is not sharable");
                                        }
                                        Tr.warning(tc, "channel.shared.warning", new Object[] { channelData.getName() });

                                    }
                                }
                                // update regions that this channel is being used in
                                ChannelZRegions.put(channelData.getName(), Integer.valueOf(entryValue | zRegion));
                            }
                        }
                    }
                }
            }
            channelDataArray[i] = channelData;
        } // end-for loop on channels

        if (okToAdd) {
            try {
                // Store values for the host and port in the chain if it is inbound.
                Map<Object, Object> chainProperties = null;
                if (FlowType.INBOUND.equals(chainType)) {
                    chainProperties = new HashMap<Object, Object>();
                    Map<Object, Object> tcpProperties = channelDataArray[0].getPropertyBag();
                    chainProperties.put(ChannelFrameworkConstants.HOST_NAME, tcpProperties.get(ChannelFrameworkConstants.HOST_NAME));
                    chainProperties.put(ChannelFrameworkConstants.PORT, tcpProperties.get(ChannelFrameworkConstants.PORT));
                    chainProperties.put(ChannelFrameworkConstants.LISTENING_PORT, tcpProperties.get(ChannelFrameworkConstants.LISTENING_PORT));
                }
                chainData = (ChainDataImpl) createChainData(chainName, chainType, channelDataArray, chainProperties);
                // Add the new chain to the framework configuration.
                this.chainDataMap.put(chainName, chainData);
                // Associate any global chain event listeners.
                for (int j = 0; j < globalChainEventListeners.size(); j++) {
                    chainData.addChainEventListener(globalChainEventListeners.get(j));
                }
            } catch (IncoherentChainException e) {
                FFDCFilter.processException(e, getClass().getName() + ".addChain", "1601", this, new Object[] { chainName, chainType, channelDataArray });
                throw e;
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "addChain");
        }
        return chainData;
    }
354518
public synchronized boolean doesChannelReferenceChain(String channelName, String chainName) {
        boolean foundRef = false;
        Chain chain = this.chainRunningMap.get(chainName);
        if (chain != null) {
            ChannelData channelsData[] = chain.getChannelsData();
            ChildChannelDataImpl childChannelData = null;
            for (int i = 0; i < channelsData.length; i++) {
                childChannelData = (ChildChannelDataImpl) channelsData[i];
                if (childChannelData.getExternalName().equals(channelName)) {
                    foundRef = true;
                    break;
                }
            }
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "doesChannelReferenceChain: channel=" + channelName + ", chain=" + chainName + ", rc=" + foundRef);
        }
        return foundRef;
    }
694323
public Channel.ID createChannel()
	{
		final int num = getNextPathNum();
		return new Channel.ID(this.paths[num], num, this.random);
	}
75448
synchronized void setChannelFactory(ChannelFactory factory) throws ChannelFactoryException {
        if (factory != null && cf != null) {
            throw new ChannelFactoryException("ChannelFactory already exists");
        }
        this.cf = factory;
    }
1009948
@Override
    public SeekableByteChannel newByteChannel(final Path path, final Set<? extends OpenOption> options,
        final FileAttribute<?>... attrs) throws IOException {
        // Precondition checks for null path done by NIO2 Files

        // Ignore FileAttributes

        // Get Archive
        final Archive<?> archive = this.getArchive(path);

        // // If overwriting isn't allowed and this path exists
        final ArchivePath archivePath = ArchivePaths.create(path.toString());

        // Open a new Channel (which is also a NamedAsset)
        final MemoryNamedAsset channel = new MemoryNamedAsset(archivePath);

        // Writing?
        if (options.contains(StandardOpenOption.CREATE) || options.contains(StandardOpenOption.CREATE_NEW)
            || options.contains(StandardOpenOption.WRITE)) {

            // Plug channel as Asset into the archive

            if (archive.contains(channel.getName())) {
                // Appending?
                if (options.contains(StandardOpenOption.APPEND)) {
                    // Read in the existing content
                    channel.position(0);
                    final InputStream in = archive.get(archivePath).getAsset().openStream();
                    this.copy(in, channel);
                    in.close();
                    // Delete the existing asset and associate the channel as the new asset
                    archive.delete(archivePath);
                    archive.add(channel);
                } else {
                    // Exception translate
                    throw new FileAlreadyExistsException(archivePath.get());
                }
            } else {
                archive.add(channel);
            }

            // Return the channel
            return channel;
        }

        // Else we're reading...

        // Get the Node
        final Node node = archive.get(archivePath);
        if (node == null) {
            throw new IllegalArgumentException(
                "No open options have been set, and cannot read a file that does not exist");
        }
        // Directory?
        final Asset asset = node.getAsset();
        if (asset == null) {
            throw new IllegalArgumentException("Cannot open a new channel to a path with existing directory: "
                + archivePath.get());
        }

        // Existing asset is read into the channel
        final InputStream in = asset.openStream();
        final SeekableByteChannel outChannel = new SeekableInMemoryByteChannel();
        this.copy(in, outChannel);
        // Set the position to 0 so it can be read from the beginning
        outChannel.position(0);
        return outChannel;

    }
733383
public static synchronized Channel getChannel() throws IOException {
        if (devnull < 0) {
            devnull = open0("/dev/null", O_RDWR);
        }

        // If we don't have the channel try to create it
        if (!haveChannel) {
            channel = createChannel();
            haveChannel = true;
        }

        // if there is a channel then do the security check before
        // returning it.
        if (channel != null) {
            checkAccess(channel);
        }
        return channel;
    }
611934
protected ItemHolder create(final SelectableChannel channel, final EventsHandler handler, final int events)
    {
        Objects.requireNonNull(channel, "Channel has to be non-null");
        return creator.create(channel, handler, events);
    }
632610
public IChannelProducer getProviderChannelFor(
            String functionalityName,
            String clientName,
            String conversationID) throws ServiceGroundingException
    {
        IChannelProducer channel = this.channelProducersHelper.getChannel(functionalityName, clientName, conversationID);
        if (null != channel && !(channel instanceof InternalChannel))
        {
            return channel;
        }

        IChannelProducer newChannel = createProviderChannelFor(functionalityName, clientName, conversationID);
        this.channelProducersHelper.addChannel(newChannel, functionalityName, clientName, conversationID);
        return newChannel;
    }
892720
def channels_create(self, name, **kwargs):
        """"""
        return self.__call_api_post('channels.create', name=name, kwargs=kwargs)
268711
@Process(actionType = InitElytron.class)
    public void init(final Dispatcher.Channel channel) {
        ResourceAddress rootAddress = ROOT_ADDRESS.resolve(statementContext);

        Operation opRealElytron = new Operation.Builder(READ_RESOURCE_OPERATION, rootAddress)
                .param(RECURSIVE, true)
                .build();

        dispatcher.execute(new DMRAction(opRealElytron), new AsyncCallback<DMRResponse>() {

            @Override
            public void onFailure(final Throwable caught) {
                channel.nack(caught);
            }

            @Override
            public void onSuccess(final DMRResponse response) {
                ModelNode result = response.get();
                if (result.isFailure()) {
                    channel.nack(result.getFailureDescription());
                } else {
                    ModelNode payload = result.get(RESULT);

                    rootNode = payload;

                    // other resources
                    populate(payload, "key-store", keyStore);
                    populate(payload, "key-manager", keyManager);
                    populate(payload, "server-ssl-context", serverSSLContext);
                    populate(payload, "client-ssl-context", clientSSLContext);
                    populate(payload, "trust-manager", trustManager);
                    populate(payload, "credential-store", credentialStore);
                    populate(payload, "filtering-key-store", filteringKeyStore);
                    populate(payload, "ldap-key-store", ldapKeyStore);
                    populate(payload, "provider-loader", providerLoader);
                    populate(payload, "aggregate-providers", aggregateProviders);
                    populate(payload, "security-domain", securityDomain);

                    populate(payload, "aggregate-principal-transformer", aggregatePrincipalTransformer);
                    populate(payload, "chained-principal-transformer", chainedPrincipalTransformer);
                    populate(payload, "constant-principal-transformer", constantPrincipalTransformer);
                    populate(payload, "custom-principal-transformer", customPrincipalTransformer);
                    populate(payload, "regex-validating-principal-transformer", regexValidatingPrincipalTransformer);
                    populate(payload, "regex-principal-transformer", regexPrincipalTransformer);

                    populate(payload, "aggregate-security-event-listener", aggregateSecurityEventListener);
                    populate(payload, "file-audit-log", fileAuditLog);
                    populate(payload, "policy", policy);
                    populate(payload, "syslog-audit-log", syslogAuditLog);
                    populate(payload, "size-rotating-file-audit-log", sizeRotatingFileAuditLog);
                    populate(payload, "periodic-rotating-file-audit-log", periodicRotatingFileAuditLog);

                    populate(payload, "dir-context", dirContext);

                    // security realms
                    populate(payload, "properties-realm", propertiesRealm);
                    populate(payload, "filesystem-realm", filesystemRealm);
                    populate(payload, "caching-realm", cachingRealm);
                    populate(payload, "jdbc-realm", jdbcRealm);
                    populate(payload, "ldap-realm", ldapRealm);
                    populate(payload, "key-store-realm", keystoreRealm);
                    populate(payload, "aggregate-realm", aggregateRealm);
                    populate(payload, "custom-modifiable-realm", customModifiableRealm);
                    populate(payload, "custom-realm", customRealm);
                    populate(payload, "identity-realm", identityRealm);
                    populate(payload, "token-realm", tokenRealm);

                    populate(payload, "mapped-regex-realm-mapper", mappedRegexRealmMapper);
                    populate(payload, "simple-regex-realm-mapper", simpleRegexRealmMapper);
                    populate(payload, "custom-realm-mapper", customRealmMapper);
                    populate(payload, "constant-realm-mapper", constantRealmMapper);

                    populate(payload, "authentication-context", authenticationContext);
                    populate(payload, "authentication-configuration", authenticationconfiguration);

                    // factories
                    populate(payload, "aggregate-http-server-mechanism-factory", aggregateHttpServerMechanismFactory);
                    populate(payload, "aggregate-sasl-server-factory", aggregateSaslServerFactory);
                    populate(payload, "configurable-http-server-mechanism-factory",
                            configurableHttpServerMechanismFactory);
                    populate(payload, "configurable-sasl-server-factory", configurableSaslServerFactory);
                    populate(payload, "custom-credential-security-factory", customCredentialSecurityFactory);
                    populate(payload, "http-authentication-factory", httpAuthenticationFactory);
                    populate(payload, "kerberos-security-factory", kerberosSecurityFactory);
                    populate(payload, "mechanism-provider-filtering-sasl-server-factory",
                            mechanismProviderFilteringSaslServerFactory);
                    populate(payload, "provider-http-server-mechanism-factory", providerHttpServerMechanismFactory);
                    populate(payload, "provider-sasl-server-factory", providerSaslServerFactory);
                    populate(payload, "sasl-authentication-factory", saslAuthenticationFactory);
                    populate(payload, "service-loader-http-server-mechanism-factory",
                            serviceLoaderHttpServerMechanismFactory);
                    populate(payload, "service-loader-sasl-server-factory", serviceLoaderSaslServerFactory);

                    // mappers and decoders
                    populate(payload, "add-prefix-role-mapper", addPrefixRoleMapper);
                    populate(payload, "add-suffix-role-mapper", addSuffixRoleMapper);
                    populate(payload, "aggregate-role-mapper", aggregateRoleMapper);
                    populate(payload, "constant-role-mapper", constantRoleMapper);
                    populate(payload, "custom-role-mapper", customRoleMapper);
                    populate(payload, "logical-role-mapper", logicalRoleMapper);

                    populate(payload, "simple-permission-mapper", simplePermissionMapper);
                    populate(payload, "logical-permission-mapper", logicalPermissionMapper);
                    populate(payload, "custom-permission-mapper", customPermissionMapper);
                    populate(payload, "constant-permission-mapper", constantPermissionMapper);

                    populate(payload, "aggregate-principal-decoder", aggregatePrincipalDecoder);
                    populate(payload, "concatenating-principal-decoder", concatenatingPrincipalDecoder);
                    populate(payload, "constant-principal-decoder", constantPrincipalDecoder);
                    populate(payload, "custom-principal-decoder", customPrincipalDecoder);
                    populate(payload, "x500-attribute-principal-decoder", x500PrincipalDecoder);
                    populate(payload, "custom-role-decoder", customRoleDecoder);
                    populate(payload, "simple-role-decoder", simpleRoleDecoder);

                    channel.ack();
                }
            }
        });
    }
1371300
@Process(actionType = CreateConnector.class)
    public void createConnector(final CreateConnector action, final Dispatcher.Channel channel) {

        final List<Property> connectors = getModelsFor(action.getAddressTemplate());
        // First create the connector, then create the security=sasl singleton and finally
        // create the sasl-policy=policy singleton
        Outcome<FunctionContext> outcome = new Outcome<FunctionContext>() {
            @Override
            public void onFailure(FunctionContext context) {
                channel.nack(context.getError());
            }

            @Override
            public void onSuccess(FunctionContext context) {
                lastModifiedInstance = action.getInstanceName();
                read(action.getAddressTemplate().getResourceType(), getModelsFor(action.getAddressTemplate()), channel);
            }
        };

        AddressTemplate securityAddress = action.getAddressTemplate().append("security=" + SASL_SINGLETON);
        AddressTemplate policyAddress = securityAddress.append("sasl-policy=" + POLICY_SINGLETON);
        new Async<FunctionContext>(new Progress.Nop()).waterfall(new FunctionContext(), outcome,
                new CreateConnectorFn(dispatcher, statementContext, action.getAddressTemplate(),
                        action.getInstanceName(), action.getNewModel()),
                new CreateSaslSingleton(dispatcher, statementContext, action.getInstanceName(), securityAddress),
                new CreateSaslSingleton(dispatcher, statementContext, action.getInstanceName(), policyAddress));
    }