销毁对应的pipeline节点,同步调用
13182
public void destory(Long channelId, Long pipelineId) {
        String path = ManagePathUtils.getPipeline(channelId, pipelineId);
        String processRootPath = ManagePathUtils.getProcessRoot(channelId, pipelineId);
        String terminRootPath = ManagePathUtils.getTerminRoot(channelId, pipelineId);
        String remedyRootPath = ManagePathUtils.getRemedyRoot(channelId, pipelineId);
        String lockRootPath = ManagePathUtils.getLockRoot(channelId, pipelineId);
        String loadLockPath = lockRootPath + "/" + ArbitrateConstants.NODE_LOCK_LOAD;
        try {
            zookeeper.deleteRecursive(loadLockPath); // 删除节点，不关心版本
            zookeeper.deleteRecursive(lockRootPath); // 删除节点，不关心版本
            zookeeper.deleteRecursive(terminRootPath); // 删除节点，不关心版本
            zookeeper.deleteRecursive(remedyRootPath); // 删除节点，不关心版本
            zookeeper.deleteRecursive(processRootPath); // 删除节点，不关心版本
            zookeeper.deleteRecursive(path); // 删除节点，不关心版本
        } catch (ZkNoNodeException e) {
            // 如果节点已经不存在，则不抛异常
            // ignore
        } catch (ZkException e) {
            throw new ArbitrateException("Pipeline_destory", pipelineId.toString(), e);
        }
    }
724829
public TerminEventData await(Long pipelineId) throws InterruptedException {
        Assert.notNull(pipelineId);
        TerminMonitor terminMonitor = ArbitrateFactory.getInstance(pipelineId, TerminMonitor.class);
        Long processId = terminMonitor.waitForProcess(); // 符合条件的processId
        if (logger.isDebugEnabled()) {
            logger.debug("## await pipeline[{}] processId[{}] is termin", pipelineId, processId);
        }

        // 根据pipelineId+processId构造对应的path
        String path = StagePathUtils.getTermin(pipelineId, processId);

        try {
            byte[] data = zookeeper.readData(path);
            return JsonUtils.unmarshalFromByte(data, TerminEventData.class);
        } catch (ZkNoNodeException e) {
            logger.error("pipeline[{}] processId[{}] is process", pipelineId, processId);
            terminMonitor.ack(processId); // modify for 2012-09-08, 发生主备切换时，await会进入死循环，针对NoNode后直接从内存队列中移除
            return await(pipelineId); // 再取下一个节点
        } catch (ZkException e) {
            throw new ArbitrateException("Termin_await", e);
        }
    }
934336
public static String getRemedyRoot(Long channelId, Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_REMEDY_ROOT,
            String.valueOf(channelId),
            String.valueOf(pipelineId));
    }
810562
def pipeline(self, transaction=True, shard_hint=None):
        '''  '''
        args_dict = {
            'connection_pool': self.connection_pool,
            'response_callbacks': self.response_callbacks,
            'transaction': transaction,
            'shard_hint': shard_hint,
            'logger': self.logger,
        }

        return StonePipeline(**args_dict)
1031941
def _pipeline(self, name):
        
        if name == self.connection_name:
            return self

        try:
            return self._pipelines[name]
        except KeyError:
            pipe = Pipeline(name=name, autoexec=True)
            self._pipelines[name] = pipe
            return pipe
652930
public Map<Long, DelayStatInfo> listTimelineDelayStat(Long pipelineId, Date start, Date end) {

        Map<Long, DelayStatInfo> delayStatInfos = new LinkedHashMap<Long, DelayStatInfo>();
        List<DelayStatDO> delayStatDOs = delayStatDao.listTimelineDelayStatsByPipelineId(pipelineId, start, end);
        int size = delayStatDOs.size();
        int k = size - 1;
        for (Long i = start.getTime(); i <= end.getTime(); i += 60 * 1000) {
            DelayStatInfo delayStatInfo = new DelayStatInfo();
            List<DelayStat> delayStats = new ArrayList<DelayStat>();
            // 取出每个时间点i以内的数据，k是一个游标，每次遍历时前面已经取过了的数据就不用再遍历了
            for (int j = k; j >= 0; --j) {
                if ((i - delayStatDOs.get(j).getGmtModified().getTime() <= 60 * 1000)
                    && (i - delayStatDOs.get(j).getGmtModified().getTime() >= 0)) {
                    delayStats.add(delayStatDOToModel(delayStatDOs.get(j)));
                    k = j - 1;
                }// 如果不满足if条件，则后面的数据也不用再遍历
                else {
                    break;
                }
            }
            if (delayStats.size() > 0) {
                delayStatInfo.setItems(delayStats);
                delayStatInfos.put(i, delayStatInfo);
            }

        }
        return delayStatInfos;
    }
390023
def _multi_rpush_pipeline(self, pipe, queue, values, bulk_size=0):
        ''' 
        '''
        cont = 0
        for value in values:
            pipe.rpush(queue, value)
            if bulk_size != 0 and cont % bulk_size == 0:
                pipe.execute()
136072
protected function runSynchronously(Pipeline $pipeline)
    {
        $results = [];

        /** @var Closure $task */
        foreach ($pipeline as $key => $task) {
            $results[$key] = $this->bash->checkResults($task());
            if (!$results[$key]) {
                break;
            }
        }

        // Update Pipeline results
        $pipeline->setResults($results);

        return $pipeline;
    }
433956
def update_pipeline(self, pipeline):
		'''
		'''
		#req sanity check
		payload = None
		if  type(pipeline) is not StreakPipeline:
			return requests.codes.bad_request, None

		payload = pipeline.to_dict(rw = True)

		try:
			uri = '/'.join([
						self.api_uri,
						self.pipelines_suffix,
						pipeline.attributes['pipelineKey']
						])
		except KeyError:
			return requests.codes.bad_request, None
	
		code, r_data = self._req('post', uri , json.dumps(payload))

		return code, r_data
54223
def _multi_lpop_pipeline(self, pipe, queue, number):
        ''' '''
        pipe.lrange(queue, 0, number - 1)
        pipe.ltrim(queue, number, -1)
1154970
public static String getProcessRoot(Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_PROCESS_ROOT, getChannelId(pipelineId),
                                    String.valueOf(pipelineId));
    }
1070985
private void configureHttp2Pipeline(ChannelPipeline pipeline) {
        pipeline.addLast(Constants.CONNECTION_HANDLER, http2ConnectionHandler);
        pipeline.addLast(Constants.HTTP2_TARGET_HANDLER, http2TargetHandler);
        pipeline.addLast(Constants.DECOMPRESSOR_HANDLER, new HttpContentDecompressor());
    }
457626
public static String getTermin(Long channelId, Long pipelineId, Long processId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_TERMIN_FORMAT,
            String.valueOf(channelId),
            String.valueOf(pipelineId),
            getProcessNode(processId));
    }
294853
def p_pipeline(p):
    ''''''
    if len(p) == 2:
        p[0] = [p[1]]
    else:
        p[0] = p[1]
        p[0].append(ast.node(kind='pipe', pipe=p[2], pos=p.lexspan(2)))
        p[0].extend(p[len(p) - 1])
110954
public static void forEachIndex(short[] input,boolean[] inputSkip,int[] output ,int outPutIndex,ForEachIndexController controller)
	{
		for(int i=0;i<input.length;i++)
		{
			if(inputSkip[i])
			{
				continue;
			}
			output[outPutIndex]=i;//输出当前位锁定值
			ComputeStatus status=controller.onOutEvent(output, outPutIndex);
			if(status.isStop())
			{
				return;
			}
			if(status.isSkip())
			{//跳过下层循环
				continue;
			}
			outPutIndex++;
			if(outPutIndex>=input.length)
			{//如果是最后一个则返回
				break;
			}
			//解锁下一个
			inputSkip[i]=true;//锁定当前位占用索引,其它位不可使用该索引
			forEachIndex(input, inputSkip, output, outPutIndex,controller);
			outPutIndex--;//回到当前
			inputSkip[i]=false;//释放锁定
		}
	}
524497
public static String getRemedyRoot(Long pipelineId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_REMEDY_ROOT, getChannelId(pipelineId),
                                    String.valueOf(pipelineId));
    }
1247634
def cli(ctx, amount, index, stage):
    """"""
    ctx.obj.say_green('Starting Streaming Pipe')
    res_pull = ctx.invoke(pull, amount=amount, index=index, stage=stage)
    res_tra = False
    if res_pull:
        # amount to transform can be less (or more)
        res_tra = ctx.invoke(
            transform, amount=amount, index=index, stage=stage)
    if res_tra:
        # amount to push can be less (or more)
        res_push = ctx.invoke(push, amount=amount, index=index, stage=stage)
    if res_pull and res_tra and res_push:
        ctx.obj.say_green('Streaming Pipe finsished')
        return True
    return False