销毁对应的channel节点,同步调用
428857
public void destory(Long channelId) {
        String path = ManagePathUtils.getChannelByChannelId(channelId);
        try {
            zookeeper.delete(path); // 删除节点，不关心版本
        } catch (ZkNoNodeException e) {
            // 如果节点已经不存在，则不抛异常
            // ignore
        } catch (ZkException e) {
            throw new ArbitrateException("Channel_destory", channelId.toString(), e);
        }
    }
1173986
def _process_channels(self,p,**params_to_override):
        
        orig_image = self._image

        for i in range(len(self._channel_data)):
            self._image = self._original_channel_data[i]
            self._channel_data[i] = self._reduced_call(**params_to_override)
        self._image = orig_image
        return self._channel_data
913638
def _refresh_channel(self):
        '''
        
        '''
        self.channel = salt.transport.client.ReqChannel.factory(self.opts)
        return self.channel
963697
public synchronized Channel getRunningChannel(String inputChannelName, Chain chain) {
        if (inputChannelName == null || chain == null) {
            return null;
        }

        Channel channel = null;
        // Ensure the chain is running.
        if (null != this.chainRunningMap.get(chain.getName())) {
            ChannelData[] channels = chain.getChannelsData();
            for (int index = 0; index < channels.length; index++) {
                if (channels[index].getExternalName().equals(inputChannelName)) {
                    channel = chain.getChannels()[index];
                    break;
                }
            }
        }
        return channel;
    }
128544
function processChannel(channel) {
    if (0 == bestAlpha) {
      return bottom[channel]
    } else {
      return bottom[channel] + (top[channel] - bottom[channel]) / bestAlpha
    }
  }
734882
private Channel doToModel(ChannelDO channelDo) {
        Channel channel = new Channel();
        try {
            channel.setId(channelDo.getId());
            channel.setName(channelDo.getName());
            channel.setDescription(channelDo.getDescription());
            channel.setStatus(arbitrateManageService.channelEvent().status(channelDo.getId()));
            channel.setParameters(channelDo.getParameters());
            channel.setGmtCreate(channelDo.getGmtCreate());
            channel.setGmtModified(channelDo.getGmtModified());
            List<Pipeline> pipelines = pipelineService.listByChannelIds(channelDo.getId());
            // 合并PipelineParameter和ChannelParameter
            SystemParameter systemParameter = systemParameterService.find();
            for (Pipeline pipeline : pipelines) {
                PipelineParameter parameter = new PipelineParameter();
                parameter.merge(systemParameter);
                parameter.merge(channel.getParameters());
                // 最后复制pipelineId参数
                parameter.merge(pipeline.getParameters());
                pipeline.setParameters(parameter);
                // pipeline.getParameters().merge(channel.getParameters());
            }
            channel.setPipelines(pipelines);
        } catch (Exception e) {
            logger.error("ERROR ## change the channel DO to Model has an exception");
            throw new ManagerException(e);
        }

        return channel;
    }
1210856
private List<NodeTask> parseNodeTask(Channel channel) {
        List<NodeTask> tasks = new ArrayList<NodeTask>();
        List<Pipeline> pipelines = channel.getPipelines();
        Long nid = configClientService.currentNode().getId();
        TaskEvent taksEvent = null;
        if (channel.getStatus().isStart()) {
            taksEvent = TaskEvent.CREATE;
        } else if (channel.getStatus().isStop()) {
            taksEvent = TaskEvent.DELETE;
        } else if (channel.getStatus().isPause()) {
            // modify by ljh at 2013-01-31 , pause状态也需要启动setl线程
            // 因为在发布的时候，restart指令不会推送指令，导致setl线程没有启动
            // return tasks;
            taksEvent = TaskEvent.CREATE;
        }
        // 处理当前最新的状态
        for (Pipeline pipeline : pipelines) {
            List<Node> sNodes = pipeline.getSelectNodes();
            for (Node node : sNodes) {
                if (nid.equals(node.getId())) {// 判断是否为当前的nid
                    NodeTask task = new NodeTask();
                    task.setPipeline(pipeline);
                    NodeTask matchTask = getMatchTask(tasks, task);
                    if (matchTask == null) {
                        matchTask = task;
                        tasks.add(task);
                    }

                    matchTask.setPipeline(pipeline);
                    matchTask.getStage().add(StageType.SELECT);
                    matchTask.getEvent().add(taksEvent);
                }
            }

            List<Node> eNodes = pipeline.getExtractNodes();
            for (Node node : eNodes) {
                if (nid.equals(node.getId())) {// 判断是否为当前的nid
                    NodeTask task = new NodeTask();
                    task.setPipeline(pipeline);
                    NodeTask matchTask = getMatchTask(tasks, task);
                    if (matchTask == null) {
                        matchTask = task;
                        tasks.add(task);
                    }

                    matchTask.getStage().add(StageType.EXTRACT);
                    matchTask.getEvent().add(taksEvent);
                }
            }

            List<Node> tlNodes = pipeline.getLoadNodes();
            for (Node node : tlNodes) {
                if (nid.equals(node.getId())) {// 判断是否为当前的nid
                    NodeTask task = new NodeTask();
                    task.setPipeline(pipeline);
                    NodeTask matchTask = getMatchTask(tasks, task);
                    if (matchTask == null) {
                        matchTask = task;
                        tasks.add(task);
                    }
                    matchTask.getStage().add(StageType.TRANSFORM);
                    matchTask.getEvent().add(taksEvent);
                    matchTask.getStage().add(StageType.LOAD);
                    matchTask.getEvent().add(taksEvent);
                }
            }
        }

        List<Long> pipelineIds = Lists.transform(channel.getPipelines(), new Function<Pipeline, Long>() {

            public Long apply(Pipeline input) {
                return input.getId();
            }
        });
        // 合并一下target中特有的记录，取一下反操作，表示要关闭
        for (NodeTask task : allTasks) {
            Pipeline pipeline = task.getPipeline();
            if (pipeline.getChannelId().equals(channel.getId()) && !pipelineIds.contains(pipeline.getId())) {
                // /是同一个channel，但对应的pipeline不在列表里
                // 处理pipeline删除
                NodeTask deletePipelineTask = new NodeTask();
                deletePipelineTask.setPipeline(pipeline);

                List<StageType> stages = task.getStage();
                List<TaskEvent> events = task.getEvent();
                for (int i = 0; i < stages.size(); i++) {
                    StageType stage = stages.get(i);
                    TaskEvent event = events.get(i);
                    if (event.isCreate()) {
                        deletePipelineTask.getStage().add(stage);
                        deletePipelineTask.getEvent().add(TaskEvent.DELETE);// 添加为关闭
                    }
                }

                tasks.add(deletePipelineTask);
            }

            if (pipelineIds.contains(pipeline.getId())) {// 在当前的channel列表中
                boolean needAdd = false;
                NodeTask matchTask = getMatchTask(tasks, task);// 找到对应的匹配
                if (matchTask == null) {
                    matchTask = new NodeTask();
                    matchTask.setPipeline(pipeline);
                    needAdd = true;
                }
                List<StageType> stages = task.getStage();
                List<TaskEvent> events = task.getEvent();
                for (int i = 0; i < stages.size(); i++) {
                    StageType stage = stages.get(i);
                    TaskEvent event = events.get(i);
                    TaskEvent matchEvent = getMatchStage(matchTask, stage);
                    if (matchEvent == null && event.isCreate()) {// 对应的stage已经被移除，触发一个DELETE操作
                        matchTask.getStage().add(stage);
                        matchTask.getEvent().add(TaskEvent.DELETE);
                    }
                }

                if (needAdd && matchTask.getStage().size() > 0) {
                    tasks.add(matchTask);
                }

            }
        }

        // 判断当前的task是否需要全部关闭
        for (NodeTask task : tasks) {
            boolean shutdown = true;
            for (TaskEvent event : task.getEvent()) {// task已为当前最新节点信息
                shutdown &= event.isDelete();
            }
            task.setShutdown(shutdown);
        }
        return tasks;
    }
815750
def join_channel(self, channel, key=None, tags=None):
        """"""
        params = [channel]
        if key:
            params.append(key)
        self.send('JOIN', params=params, tags=tags)
328232
function(channel){
    
        var tokenInfo   = this._tokenData,
            reader      = this._reader,
            value,
            i           =0,
            len         = tokenInfo.length,
            found       = false,
            token,
            info;
            
        //check the lookahead buffer first
        if (this._lt.length && this._ltIndex >= 0 && this._ltIndex < this._lt.length){  
                           
            i++;
            this._token = this._lt[this._ltIndex++];
            info = tokenInfo[this._token.type];
            
            //obey channels logic
            while((info.channel !== undefined && channel !== info.channel) &&
                    this._ltIndex < this._lt.length){
                this._token = this._lt[this._ltIndex++];
                info = tokenInfo[this._token.type];
                i++;
            }
            
            //here be dragons
            if ((info.channel === undefined || channel === info.channel) &&
                    this._ltIndex <= this._lt.length){
                this._ltIndexCache.push(i);
                return this._token.type;
            }
        }
        
        //call token retriever method
        token = this._getToken();

        //if it should be hidden, don't save a token
        if (token.type > -1 && !tokenInfo[token.type].hide){
                     
            //apply token channel
            token.channel = tokenInfo[token.type].channel;
         
            //save for later
            this._token = token;
            this._lt.push(token);

            //save space that will be moved (must be done before array is truncated)
            this._ltIndexCache.push(this._lt.length - this._ltIndex + i);  
        
            //keep the buffer under 5 items
            if (this._lt.length > 5){
                this._lt.shift();                
            }
            
            //also keep the shift buffer under 5 items
            if (this._ltIndexCache.length > 5){
                this._ltIndexCache.shift();
            }
                
            //update lookahead index
            this._ltIndex = this._lt.length;
        }
            
        /*
         * Skip to the next token if:
         * 1. The token type is marked as hidden.
         * 2. The token type has a channel specified and it isn't the current channel.
         */
        info = tokenInfo[token.type];
        if (info && 
                (info.hide || 
                (info.channel !== undefined && channel !== info.channel))){
            return this.get(channel);
        } else {
            //return just the type
            return token.type;
        }
    }
394207
def connect_channels(self, channels):
        """"""
        self.log.info(f"Connecting to channels...")
        for chan in channels:
            chan.connect(self.sock)
            self.log.info(f"\t{chan.channel}")
663832
private void handle(SelectionKey key) {
		// 有客户端接入此服务端
		if (key.isAcceptable()) {
			// 获取通道 转化为要处理的类型
			final ServerSocketChannel server = (ServerSocketChannel) key.channel();
			SocketChannel socketChannel;
			try {
				// 获取连接到此服务器的客户端通道
				socketChannel = server.accept();
			} catch (IOException e) {
				throw new IORuntimeException(e);
			}

			// SocketChannel通道的可读事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.READ);
		}

		// 读事件就绪
		if (key.isReadable()) {
			final SocketChannel socketChannel = (SocketChannel) key.channel();
			read(socketChannel);

			// SocketChannel通道的可写事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.WRITE);
		}

		// 写事件就绪
		if (key.isWritable()) {
			final SocketChannel socketChannel = (SocketChannel) key.channel();
			write(socketChannel);
			// SocketChannel通道的可读事件注册到Selector中
			registerChannel(selector, socketChannel, Operation.READ);
		}
	}
934494
public boolean stop(Long channelId, boolean needTermin) {
        // stop优先级高于pause
        updateStatus(channelId, ChannelStatus.STOP);

        boolean result = !needTermin;

        if (needTermin) {
            try {
                result |= termin(channelId, TerminType.SHUTDOWN);
            } catch (Throwable e) {
                updateStatus(channelId, ChannelStatus.STOP); // 出错了，直接挂起
                throw new ArbitrateException(e);
            }
        }

        return result;
    }
1183073
public void addChannel(WebSocketChannel channel) {
        Objects.requireNonNull(channel, Required.CHANNEL.toString());

        final String url = RequestUtils.getWebSocketURL(channel);
        Set<WebSocketChannel> channels = getChannels(url);
        if (channels == null) {
            channels = new HashSet<>();
            channels.add(channel);
        } else {
            channels.add(channel);
        }
        setChannels(url, channels);
    }
641307
public boolean pause(Long channelId, boolean needTermin) {
        ChannelStatus currstatus = status(channelId);
        boolean status = false;
        boolean result = !needTermin;
        if (currstatus.isStart()) { // stop的优先级高于pause，这里只针对start状态进行状态更新
            updateStatus(channelId, ChannelStatus.PAUSE);
            status = true; // 避免stop时发生rollback报警
        }

        if (needTermin) {
            try {
                // 调用termin进行关闭
                result |= termin(channelId, TerminType.ROLLBACK);
            } catch (Throwable e) {
                updateStatus(channelId, ChannelStatus.PAUSE); // 出错了，直接挂起
                throw new ArbitrateException(e);
            }
        }

        return result && status;
    }
1348593
public static String getChannelByChannelId(Long channelId) {
        // 根据channelId 构造path
        return MessageFormat.format(ArbitrateConstants.NODE_CHANNEL_FORMAT, String.valueOf(channelId));
    }
1253420
def bind_to_channel(self):
        
        if self.channel.code_name != self.user.prv_exchange:
            channel = self._connect_mq()
            channel.exchange_bind(source=self.channel.code_name, destination=self.user.prv_exchange)
453108
def channel_info(self, chan):
        """"""
        info=fluid_synth_channel_info_t()
        fluid_synth_get_channel_info(self.synth, chan, byref(info))
        return (info.sfont_id, info.bank, info.program, info.name)