in src/python/pants/backend/jvm/tasks/jvm_compile/rsc/rsc_compile.py [0:0]
def register_extra_products_from_contexts(self, targets, compile_contexts):
super(RscCompile, self).register_extra_products_from_contexts(targets, compile_contexts)
def pathglob_for(filename):
return PathGlobsAndRoot(
PathGlobs(
(fast_relpath_optional(filename, get_buildroot()),)),
text_type(get_buildroot()))
def to_classpath_entries(paths, scheduler):
# list of path ->
# list of (path, optional<digest>) ->
path_and_digests = [(p, Digest.load(os.path.dirname(p))) for p in paths]
# partition: list of path, list of tuples
paths_without_digests = [p for (p, d) in path_and_digests if not d]
if paths_without_digests:
self.context.log.debug('Expected to find digests for {}, capturing them.'
.format(paths_without_digests))
paths_with_digests = [(p, d) for (p, d) in path_and_digests if d]
# list of path -> list path, captured snapshot -> list of path with digest
snapshots = scheduler.capture_snapshots(tuple(pathglob_for(p) for p in paths_without_digests))
captured_paths_and_digests = [(p, s.directory_digest)
for (p, s) in zip(paths_without_digests, snapshots)]
# merge and classpath ify
return [ClasspathEntry(p, d) for (p, d) in paths_with_digests + captured_paths_and_digests]
def confify(entries):
return [(conf, e) for e in entries for conf in self._confs]
# Ensure that the jar/rsc jar is on the rsc_classpath.
for target in targets:
rsc_cc, compile_cc = compile_contexts[target]
if rsc_cc.workflow is not None:
cp_entries = rsc_cc.workflow.resolve_for_enum_variant({
'zinc-only': lambda : confify([compile_cc.jar_file]),
'rsc-then-zinc': lambda : confify(
to_classpath_entries([rsc_cc.rsc_jar_file], self.context._scheduler)),
})()
self.context.products.get_data('rsc_classpath').add_for_target(
target,
cp_entries)