指定した内容のTupleを送信する。

@param fileLines ファイルの内容
@param collector Collector
340884
protected void emitTuples(List<String> fileLines, TridentCollector collector)
    {
        for (String targetLine : fileLines)
        {
            collector.emit(new Values(targetLine));
        }
    }
1338619
public static Path appendLinesAndClose(String filePath, Charset charset,
            Collection<String> lineCollection) {
        return appendLineStreamAndClose(filePath, charset,
                lineCollection.stream());
    }
515489
def filter(self,
               predicate: Callable[[FileLine], 'FileLineSet']
               ) -> 'FileLineSet':
        
        filtered = [fileline for fileline in self if predicate(fileline)]
        return FileLineSet.from_list(filtered)
19965
public static Path appendLinesAndClose(String filePath,
            Collection<String> lineCollection) {
        return appendLinesAndClose(filePath, UTF_8, lineCollection);
    }
204069
public void setWithTotal() throws IOException {
    if (collectorType.equals(DataCollector.COLLECTOR_TYPE_LIST)) {
      if (segmentName != null) {
        throw new IOException("can't get total with segmentRegistration");
      } else {
        withTotal = true;
      }
    } else {
      throw new IOException(
          "can't get total for dataCollector of type " + collectorType);
    }
  }
67693
private static void getFilesByNameRecursive(Collection<File> filesList, File path, String name)
    {
        Optional.ofNullable(path.listFiles()).ifPresent(files -> Arrays.asList(files).forEach(content ->
        {
            if (content.isFile() && content.getName().equals(name))
            {
                filesList.add(content);
            }
            else if (content.isDirectory())
            {
                getFilesByNameRecursive(filesList, content, name);
            }
        }));
    }
1333918
public static void addFilesRecursively(File dir, Collection<File> fileCollection) {
		log.debug("Adding files from {}", dir.getPath());
		File[] files = dir.listFiles();
		for (File file : files) {
			if (file.isDirectory())
				addFilesRecursively(file, fileCollection);
			else {
				fileCollection.add(file);
			}
		}
	}
788310
public static void collectStats(final StatsCollector collector) {
    collector.record("connectionmgr.connections", channels.size(), "type=open");
    collector.record("connectionmgr.connections", connections_rejected,
        "type=rejected");
    collector.record("connectionmgr.connections", connections_established, 
        "type=total");
    collector.record("connectionmgr.exceptions", exceptions_closed, 
        "type=closed");
    collector.record("connectionmgr.exceptions", exceptions_reset, 
        "type=reset");
    collector.record("connectionmgr.exceptions", exceptions_timeout, 
        "type=timeout");
    collector.record("connectionmgr.exceptions", exceptions_unknown, 
        "type=unknown");
  }
790356
@Override
	protected List<File> resolve(String fqtn) throws Exception {
		StringTokenizer tokenizer = new StringTokenizer(fqtn, ":", false);
		String groupId = tokenizer.nextToken();
		String templateId = tokenizer.nextToken();
		String version = tokenizer.nextToken();

		Dependency dependency = new Dependency(new DefaultArtifact(groupId + ':' + templateId + ':' + version), null);

		CollectRequest collectRequest = new CollectRequest();
		collectRequest.setRoot(dependency);
		collectRequest.addRepository(local);
		collectRequest.addRepository(central);

		RepositorySystemSession session = newSession(repoSystem);

		DependencyNode node = repoSystem.collectDependencies(session, collectRequest).getRoot();
		DependencyRequest dependencyRequest = new DependencyRequest(node, null);

		repoSystem.resolveDependencies(session, dependencyRequest);

		PreorderNodeListGenerator nlg = new PreorderNodeListGenerator();
		node.accept(nlg);
		return nlg.getFiles();
	}
15134
@Override
	public void collect(FeatureCollector collector) {
		logBanner("Starting...");
		Long startTime = System.currentTimeMillis();
		
		Set<Project> projects = featureService.getProjectsToUpdate(collector.getId());
		List<Future<UpdateResult>> updateIssuesFutures = updateIssuesForProjects(collector, projects);
		logResults(updateIssuesFutures, startTime);
	}
489966
public static FileCollection list(FileListParams params, RequestOptions options)
      throws StripeException {
    return requestCollection(classUrl(File.class), params, FileCollection.class, options);
  }
1054207
public void collectTfIdfTerms(boolean collectTfIdfTerms) {
    if (this.collectTfIdfTerms != collectTfIdfTerms) {
      this.collectTfIdfTerms = collectTfIdfTerms;
      changed = true;
      if (tfidfCalculator != null) {
        tfidfCalculator.enableTermCollection(collectTfIdfTerms);
      }
    }
  }
881082
@SuppressWarnings("PMD.AvoidDeeplyNestedIfStmts") // agreed, fixme
    private void clean(Collector collector) {
        Set<ObjectId> uniqueIDs = new HashSet<>();
        /*
          Logic: For each component, retrieve the collector item list of the type SCM.
          Store their IDs in a unique set ONLY if their collector IDs match with GitHub collectors ID.
         */
        for (com.capitalone.dashboard.model.Component comp : dbComponentRepository.findAll()) {
            if (!MapUtils.isEmpty(comp.getCollectorItems())) {
                List<CollectorItem> itemList = comp.getCollectorItems().get(CollectorType.SCM);
                if (itemList != null) {
                    itemList.stream().filter(ci -> ci != null && Objects.equals(ci.getCollectorId(), collector.getId())).map(BaseModel::getId).forEach(uniqueIDs::add);
                }
            }
        }
        /*
          Logic: Get all the collector items from the collector_item collection for this collector.
          If their id is in the unique set (above), keep them enabled; else, disable them.
         */
        List<GitHubRepo> repoList = new ArrayList<>();
        Set<ObjectId> gitID = new HashSet<>();
        gitID.add(collector.getId());
        gitHubRepoRepository.findByCollectorIdIn(gitID).stream().filter(Objects::nonNull).forEach(repo -> {
            if (repo.isPushed()) {return;}
            repo.setEnabled(uniqueIDs.contains(repo.getId()));
            repoList.add(repo);
        });
        gitHubRepoRepository.save(repoList);
    }
216537
private RemoteIterator<LocatedFileStatus> iteratorListing(final String src)
  throws IOException {
    return new RemoteIterator<LocatedFileStatus>() {
      private LocatedDirectoryListing thisListing;
      private int i;

      { // initializer
        // fetch the first batch of entries in the directory

        thisListing = namenode.getLocatedPartialListing(
            src, HdfsFileStatus.EMPTY_NAME);
        if (thisListing == null) { // the directory does not exist
          throw new FileNotFoundException("File " + src + " does not exist.");
        }
      }

      @Override
      public boolean hasNext() throws IOException {
        if (i>=thisListing.getPartialListing().length
            && thisListing.hasMore()) {
          // current listing is exhausted & fetch a new listing
          thisListing = namenode.getLocatedPartialListing(
              src, thisListing.getLastName());
          if (thisListing == null) {
            throw new FileNotFoundException("File " + src + " does not exist.");
          }
          i = 0;
        }
        return i < thisListing.getPartialListing().length;
      }

      @Override
      public LocatedFileStatus next() throws IOException {
        if (!hasNext()) {
          throw new java.util.NoSuchElementException("No more entry in " + src);
        }
        return HdfsFileStatus.toLocatedFileStatus(thisListing.getPartialListing()[i],
            thisListing.getBlockLocations()[i++], src);
      }
    };

  }
478528
public static void collectStats(final StatsCollector collector) {
    collector.record("rpc.forbidden", telnet_requests_forbidden, "type=telnet");
    collector.record("rpc.unauthorized", telnet_requests_unauthorized, "type=telnet");
    collector.record("rpc.forbidden", http_requests_forbidden, "type=http");
    collector.record("rpc.unauthorized", http_requests_unauthorized, "type=http");
    collector.record("rpc.received", http_requests, "type=put");
    collector.record("rpc.errors", hbase_errors, "type=hbase_errors");
    collector.record("rpc.errors", invalid_values, "type=invalid_values");
    collector.record("rpc.errors", illegal_arguments, "type=illegal_arguments");
    collector.record("rpc.errors", unknown_metrics, "type=unknown_metrics");
    collector.record("rpc.errors", writes_blocked, "type=socket_writes_blocked");
  }
754283
public static <T extends Collection<String>> T readLines(InputStream in, Charset charset, T collection) throws IORuntimeException {
		return readLines(getReader(in, charset), collection);
	}
671950
public static ArrayList<String> getFilteredLinesFromFile(final String fileName, final String filter)
      throws IOException {
    return getFilteredLinesFromFile(fileName, filter, null, null);
  }