Return the full aggregation pipeline for this ChangeStream.
775958
def _full_pipeline(self):
        """"""
        options = self._pipeline_options()
        full_pipeline = [{'$changeStream': options}]
        full_pipeline.extend(self._pipeline)
        return full_pipeline
1254727
function (cmd) {
    var aggregateId = dotty.get(cmd, this.definitions.command.aggregateId);
    var aggregateName = this.aggregate.name;
    var contextName = this.aggregate.context.name;

    var toLoad = [{
      context: contextName,
      aggregate: aggregateName,
      aggregateId: aggregateId
    }];

    if (_.isFunction(this.getStreamsInfo)) {
      toLoad = this.getStreamsInfo(cmd);
      if (!_.isArray(toLoad)) {
        toLoad = [toLoad];
      }
      toLoad.forEach(function (l) {
        l.aggregateId = l.aggregateId || aggregateId;
      });
    }

    if (toLoad[0].context !== contextName || toLoad[0].aggregate !== aggregateName) {
      throw new Error('First stream to load has to be the new one!');
    }

    return toLoad;
  }
1064204
def _run_aggregation_cmd(self, session, explicit_session):
        
        read_preference = self._target._read_preference_for(session)
        client = self._database.client

        def _cmd(session, server, sock_info, slave_ok):
            pipeline = self._full_pipeline()
            cmd = SON([("aggregate", self._aggregation_target),
                       ("pipeline", pipeline),
                       ("cursor", {})])

            result = sock_info.command(
                self._database.name,
                cmd,
                slave_ok,
                read_preference,
                self._target.codec_options,
                parse_write_concern_error=True,
                read_concern=self._target.read_concern,
                collation=self._collation,
                session=session,
                client=self._database.client)

            cursor = result["cursor"]

            if (self._start_at_operation_time is None and
                self._resume_token is None and
                cursor.get("_id") is None and
                sock_info.max_wire_version >= 7):
                self._start_at_operation_time = result["operationTime"]

            ns = cursor["ns"]
            _, collname = ns.split(".", 1)
            aggregation_collection = self._database.get_collection(
                collname, codec_options=self._target.codec_options,
                read_preference=read_preference,
                write_concern=self._target.write_concern,
                read_concern=self._target.read_concern
            )

            return CommandCursor(
                aggregation_collection, cursor, sock_info.address,
                batch_size=self._batch_size or 0,
                max_await_time_ms=self._max_await_time_ms,
                session=session, explicit_session=explicit_session)

        return client._retryable_read(_cmd, read_preference, session)
664401
function (aggregate, streams, clb) {
        agg = aggregate; // save it temporary so we can use it in the callback

        debug('');
        var err = self.verifyAggregate(aggregate, cmd);
        if (err) {
          return clb(err);
        }
        clb(null, aggregate, streams);
      }
161284
def update_pipeline_stage(self, stage):
		'''
		'''
		#req sanity check
		payload = None
		if  type(stage) is not StreakStage:
			return requests.codes.bad_request, None

		payload = stage.to_dict(rw = True)
	
		#print(new_pl.attributes)
		#print(new_pl.to_dict())
		#raw_input()
		try:
			uri = '/'.join([self.api_uri,
							self.pipelines_suffix,
							stage.attributes['pipelineKey'],
							self.stages_suffix,
							stage.attributes['key']
							])
		except KeyError:
			return requests.codes.bad_request, None
	
		code, data = self._req('post', uri , json.dumps(payload))
		
		return code, data
1240678
def get_diff_amounts(self):
        
        diffs = []

        last_commit = None
        for commit in self.repo.iter_commits():
            if last_commit is not None:
                diff = self.get_diff(commit.hexsha, last_commit.hexsha)
                total_changed = diff[Diff.ADD] + diff[Diff.DEL]
                diffs.append(total_changed)

            last_commit = commit

        return diffs
33836
public static String getTransformStage(Long pipelineId, Long processId) {
        return getProcess(pipelineId, processId) + "/" + ArbitrateConstants.NODE_TRANSFORMED;
    }
289987
def aggregate(cls, pipeline=None, **kwargs):
        
        return list(cls.collection.aggregate(pipeline or [], **kwargs))
183251
def _get(self, name, interval, config, timestamp, **kws):
    '''
    
    '''
    i_bucket, r_bucket, i_key, r_key = self._calc_keys(config, name, timestamp)
    fetch = kws.get('fetch') or self._type_get
    process_row = kws.get('process_row') or self._process_row

    rval = OrderedDict()
    if config['coarse']:
      data = process_row( fetch(self._client, i_key) )
      rval[ config['i_calc'].from_bucket(i_bucket) ] = data
    else:
      # First fetch all of the resolution buckets for this set.
      resolution_buckets = sorted(map(int,self._client.smembers(i_key)))

      # Create a pipe and go fetch all the data for each.
      # TODO: turn off transactions here?
      pipe = self._client.pipeline(transaction=False)
      for bucket in resolution_buckets:
        r_key = '%s:%s'%(i_key, bucket)   # TODO: make this the "resolution_bucket" closure?
        fetch(pipe, r_key)
      res = pipe.execute()

      for idx,data in enumerate(res):
        data = process_row(data)
        rval[ config['r_calc'].from_bucket(resolution_buckets[idx]) ] = data

    return rval
1276465
def pipeline_summary(data):
    
    data = utils.to_single_data(data)
    work_bam = dd.get_align_bam(data) or dd.get_work_bam(data)
    if not work_bam or not work_bam.endswith(".bam"):
        work_bam = None
    if dd.get_ref_file(data):
        if work_bam or (tz.get_in(["config", "algorithm", "kraken"], data)):  # kraken doesn't need bam
            logger.info("QC: %s %s" % (dd.get_sample_name(data), ", ".join(dd.get_algorithm_qc(data))))
            work_data = cwlutils.unpack_tarballs(utils.deepish_copy(data), data)
            data["summary"] = _run_qc_tools(work_bam, work_data)
            if (len(dd.get_algorithm_qc(data)) == 1 and "output_cwl_keys" in data):
                data["summary"]["qc"] = data["summary"]["qc"].get(dd.get_algorithm_qc(data)[0])
    return [[data]]
75500
def streams(self):
        

        if not self.get_objects_by_type('stream'):
            tpld_ids = []
            for index in self.get_attribute('ps_indices').split():
                stream = XenaStream(parent=self, index='{}/{}'.format(self.index, index))
                tpld_ids.append(stream.get_attribute('ps_tpldid'))
            if tpld_ids:
                XenaStream.next_tpld_id = max([XenaStream.next_tpld_id] + [int(t) for t in tpld_ids]) + 1
        return {s.id: s for s in self.get_objects_by_type('stream')}
612708
function(self) {
  if (self.resumeToken) {
    self.options.resumeAfter = self.resumeToken;
  }

  var changeStreamCursor = buildChangeStreamAggregationCommand(self);

  /**
   * Fired for each new matching change in the specified namespace. Attaching a `change`
   * event listener to a Change Stream will switch the stream into flowing mode. Data will
   * then be passed as soon as it is available.
   *
   * @event ChangeStream#change
   * @type {object}
   */
  if (self.listenerCount('change') > 0) {
    changeStreamCursor.on('data', function(change) {
      processNewChange({ changeStream: self, change, eventEmitter: true });
    });
  }

  /**
   * Change stream close event
   *
   * @event ChangeStream#close
   * @type {null}
   */
  changeStreamCursor.on('close', function() {
    self.emit('close');
  });

  /**
   * Change stream end event
   *
   * @event ChangeStream#end
   * @type {null}
   */
  changeStreamCursor.on('end', function() {
    self.emit('end');
  });

  /**
   * Fired when the stream encounters an error.
   *
   * @event ChangeStream#error
   * @type {Error}
   */
  changeStreamCursor.on('error', function(error) {
    processNewChange({ changeStream: self, error, eventEmitter: true });
  });

  if (self.pipeDestinations) {
    const cursorStream = changeStreamCursor.stream(self.streamOptions);
    for (let pipeDestination in self.pipeDestinations) {
      cursorStream.pipe(pipeDestination);
    }
  }

  return changeStreamCursor;
}
854420
def get_all_changes(self, *args, **kwds):
        """"""
        result = []
        for project, refactoring in zip(self.projects, self.refactorings):
            args, kwds = self._resources_for_args(project, args, kwds)
            result.append((project, refactoring.get_changes(*args, **kwds)))
        return result
1050545
protected static RedisFuture<Long> aggregateAsync(Map<?, ? extends CompletionStage<Long>> executions) {

        return new PipelinedRedisFuture<>(executions, objectPipelinedRedisFuture -> {
            AtomicLong result = new AtomicLong();
            for (CompletionStage<Long> future : executions.values()) {
                Long value = execute(() -> future.toCompletableFuture().get());
                if (value != null) {
                    result.getAndAdd(value);
                }
            }

            return result.get();
        });
    }
569114
public function assemblePipeline() {
		$aggregate = [];
		
		// Conditions
		if (! empty($this->_conditions)) {
			$aggregate[] = [self::MONGO_PIPE_MATCH => $this->_conditions];
		}
		
		// Group
		if (! empty($this->_groupDefinition)) {
			$group = $this->_groupDefinition;
			
			$group[MongoCollection::DEFAULT_PRIMARY] = ($this->_groupBy === null)
				? null
				: '$' . $this->_groupBy;
			
			$aggregate[] = [self::MONGO_PIPE_GROUP => $group];
		}
		
		// Sort
		if (! empty($this->_sort)) {
			$aggregate[] = [self::MONGO_PIPE_SORT => $this->_sort];
		}
		
		// Offset
		if ($this->_offset !== null) {
			$aggregate[] = [self::MONGO_PIPE_OFFSET => $this->_offset];
		}
		
		// Limit
		if ($this->_limit !== null) {
			$aggregate[] = [self::MONGO_PIPE_LIMIT => $this->_limit];
		}
		
		return $aggregate;
	}
922495
def get_upload_pipeline(in_fd, out_fd, rate_limit=None,
                        gpg_key=None, lzop=True):
    
    commands = []
    if rate_limit is not None:
        commands.append(PipeViewerRateLimitFilter(rate_limit))
    if lzop:
        commands.append(LZOCompressionFilter())

    if gpg_key is not None:
        commands.append(GPGEncryptionFilter(gpg_key))

    return Pipeline(commands, in_fd, out_fd)
221684
private CompletableFuture<List<EpochRecord>> getEpochRecords(String scope, String stream, int epoch, OperationContext context) {
        List<CompletableFuture<EpochRecord>> list = new ArrayList<>();
        list.add(streamMetadataStore.getEpoch(scope, stream, epoch, context, executor));
        list.add(streamMetadataStore.getActiveEpoch(scope, stream, context, true, executor));
        return Futures.allOfWithResults(list);
    }