text:
package com.streamsets.datacollector.stagelibrary;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Splitter;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import com.streamsets.datacollector.classpath.ClasspathValidator;
import com.streamsets.datacollector.classpath.ClasspathValidatorResult;
import com.streamsets.datacollector.config.CredentialStoreDefinition;
import com.streamsets.datacollector.config.ErrorHandlingChooserValues;
import com.streamsets.datacollector.config.InterceptorDefinition;
import com.streamsets.datacollector.config.LineagePublisherDefinition;
import com.streamsets.datacollector.config.PipelineDefinition;
import com.streamsets.datacollector.config.PipelineFragmentDefinition;
import com.streamsets.datacollector.config.PipelineLifecycleStageChooserValues;
import com.streamsets.datacollector.config.PipelineRulesDefinition;
import com.streamsets.datacollector.config.PipelineTestStageChooserValues;
import com.streamsets.datacollector.config.PrivateClassLoaderDefinition;
import com.streamsets.datacollector.config.ServiceDefinition;
import com.streamsets.datacollector.config.ServiceDependencyDefinition;
import com.streamsets.datacollector.config.StageDefinition;
import com.streamsets.datacollector.config.StageLibraryDefinition;
import com.streamsets.datacollector.config.StageLibraryDelegateDefinitition;
import com.streamsets.datacollector.config.StatsTargetChooserValues;
import com.streamsets.datacollector.definition.CredentialStoreDefinitionExtractor;
import com.streamsets.datacollector.definition.InterceptorDefinitionExtractor;
import com.streamsets.datacollector.definition.LineagePublisherDefinitionExtractor;
import com.streamsets.datacollector.definition.ServiceDefinitionExtractor;
import com.streamsets.datacollector.definition.StageDefinitionExtractor;
import com.streamsets.datacollector.definition.StageLibraryDefinitionExtractor;
import com.streamsets.datacollector.definition.StageLibraryDelegateDefinitionExtractor;
import com.streamsets.datacollector.el.RuntimeEL;
import com.streamsets.datacollector.json.JsonMapperImpl;
import com.streamsets.datacollector.json.ObjectMapperFactory;
import com.streamsets.datacollector.main.BuildInfo;
import com.streamsets.datacollector.main.RuntimeInfo;
import com.streamsets.datacollector.main.SdcConfiguration;
import com.streamsets.datacollector.restapi.bean.RepositoryManifestJson;
import com.streamsets.datacollector.restapi.bean.StageInfoJson;
import com.streamsets.datacollector.restapi.bean.StageLibrariesJson;
import com.streamsets.datacollector.restapi.bean.StageLibraryManifestJson;
import com.streamsets.datacollector.runner.ServiceRuntime;
import com.streamsets.datacollector.runner.StageLibraryDelegateRuntime;
import com.streamsets.datacollector.task.AbstractTask;
import com.streamsets.datacollector.util.Configuration;
import com.streamsets.datacollector.util.Version;
import com.streamsets.pipeline.SDCClassLoader;
import com.streamsets.pipeline.api.ext.DataCollectorServices;
import com.streamsets.pipeline.api.ext.json.JsonMapper;
import com.streamsets.pipeline.api.impl.LocaleInContext;
import com.streamsets.pipeline.api.impl.Utils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.inject.Inject;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.core.Response;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

public class ClassLoaderStageLibraryTask extends AbstractTask implements StageLibraryTask {
  public static final String MAX_PRIVATE_STAGE_CLASS_LOADERS_KEY = "max.stage.private.classloaders";
  public static final int MAX_PRIVATE_STAGE_CLASS_LOADERS_DEFAULT = 50;

  public static final String IGNORE_STAGE_DEFINITIONS = "ignore.stage.definitions";
  public static final String JAVA_UNSUPPORTED_REGEXP = "java.unsupported.regexp";
  public static final String MIN_SDC_VERSION = "min.sdc.version";

  private static final String CONFIG_LIBRARY_ALIAS_PREFIX = "library.alias.";
  private static final String CONFIG_STAGE_ALIAS_PREFIX = "stage.alias.";

  private static final String PROPERTIES_CP_WHITELIST = "data-collector-classpath-whitelist.properties";

  private static final String CONFIG_CP_VALIDATION = "stagelibs.classpath.validation.enable";
  private static final boolean DEFAULT_CP_VALIDATION = true;

  private static final String CONFIG_CP_VALIDATION_RESULT = "stagelibs.classpath.validation.terminate";
  private static final boolean DEFAULT_CP_VALIDATION_RESULT = false;

  private static final String DEFAULT_REQUIRED_STAGELIBS = "";

  private static final String NIGHTLY_URL = "http://nightly.streamsets.com/datacollector/";
  private static final String ARCHIVES_URL = "http://archives.streamsets.com/datacollector/";
  private static final String LATEST = "latest";
  private static final String SNAPSHOT = "-SNAPSHOT";
  private static final String TARBALL_PATH = "/tarball/";
  private static final String ENTERPRISE_PATH = "enterprise/";
  private static final String CONFIG_PACKAGE_MANAGER_REPOSITORY_LINKS = "package.manager.repository.links";
  private static final String REPOSITORY_MANIFEST_JSON_PATH = "repository.manifest.json";
  private static final String ADDITIONAL = "additional";

  private static final Logger LOG = LoggerFactory.getLogger(ClassLoaderStageLibraryTask.class);

  private final RuntimeInfo runtimeInfo;
  private final BuildInfo buildInfo;
  private final Map<String,String> libraryNameAliases;
  private final Map<String,String> stageNameAliases;
  private final Configuration configuration;
  private List<? extends ClassLoader> stageClassLoaders;
  private List<StageLibraryDefinition> stageLibraries;
  private Map<String, StageDefinition> stageMap;
  private List<StageDefinition> stageList;
  private List<LineagePublisherDefinition> lineagePublisherDefinitions;
  private Map<String, LineagePublisherDefinition> lineagePublisherDefinitionMap;
  private List<CredentialStoreDefinition> credentialStoreDefinitions;
  private LoadingCache<Locale, List<StageDefinition>> localizedStageList;
  private List<ServiceDefinition> serviceList;
  private Map<Class, ServiceDefinition> serviceMap;
  private List<InterceptorDefinition> interceptorList;
  private List<StageLibraryDelegateDefinitition> delegateList;
  private Map<String, StageLibraryDelegateDefinitition> delegateMap;
  private ObjectMapper json;
  private KeyedObjectPool<String, ClassLoader> privateClassLoaderPool;

  private List<RepositoryManifestJson> repositoryManifestList = null;

  @Inject
  public ClassLoaderStageLibraryTask(RuntimeInfo runtimeInfo, BuildInfo buildInfo, Configuration configuration) {
    super("stageLibrary");
    this.runtimeInfo = runtimeInfo;
    this.buildInfo = buildInfo;
    this.configuration = configuration;
    Map<String, String> aliases = new HashMap<>();
    for (Map.Entry<String,String> entry
        : configuration.getSubSetConfiguration(CONFIG_LIBRARY_ALIAS_PREFIX).getValues().entrySet()) {
      aliases.put(entry.getKey().substring(CONFIG_LIBRARY_ALIAS_PREFIX.length()), entry.getValue());
    }
    libraryNameAliases = ImmutableMap.copyOf(aliases);
    aliases.clear();
    for (Map.Entry<String,String> entry
      : configuration.getSubSetConfiguration(CONFIG_STAGE_ALIAS_PREFIX).getValues().entrySet()) {
      aliases.put(entry.getKey().substring(CONFIG_STAGE_ALIAS_PREFIX.length()), entry.getValue());
    }
    stageNameAliases = ImmutableMap.copyOf(aliases);
  }

  private Method duplicateClassLoaderMethod;
  private Method getClassLoaderKeyMethod;
  private Method isPrivateClassLoaderMethod;

  private void resolveClassLoaderMethods(ClassLoader cl) {
    if (cl.getClass().getSimpleName().equals("SDCClassLoader")) {
      try {
        duplicateClassLoaderMethod = cl.getClass().getMethod("duplicateStageClassLoader");
        getClassLoaderKeyMethod = cl.getClass().getMethod("getName");
        isPrivateClassLoaderMethod = cl.getClass().getMethod("isPrivate");
      } catch (Exception ex) {
        throw new Error(ex);
      }
    } else {
      LOG.warn("No SDCClassLoaders available, there is no class isolation");
    }
  }

  @SuppressWarnings("unchecked")
  private <T> T invoke(Method method, ClassLoader cl, Class<T> returnType) {
    try {
      return (T) method.invoke(cl);
    } catch (Exception ex) {
      throw new Error(ex);
    }
  }

  private ClassLoader duplicateClassLoader(ClassLoader cl) {
    return (duplicateClassLoaderMethod == null) ? cl : invoke(duplicateClassLoaderMethod, cl, ClassLoader.class);
  }

  private String getClassLoaderKey(ClassLoader cl) {
    return (getClassLoaderKeyMethod == null) ? "key" : invoke(getClassLoaderKeyMethod, cl, String.class);
  }

  private boolean isPrivateClassLoader(ClassLoader cl) {
    if (cl != getClass().getClassLoader()) { // if we are the container CL we are not private for sure
      return (isPrivateClassLoaderMethod == null) ? false : invoke(isPrivateClassLoaderMethod, cl, Boolean.class);
    } else {
      return  false;
    }
  }

  private class ClassLoaderFactory extends BaseKeyedPooledObjectFactory<String, ClassLoader> {
    private final Map<String, ClassLoader> classLoaderMap;

    public ClassLoaderFactory(List<? extends ClassLoader> classLoaders) {
      classLoaderMap = new HashMap<>();
      for (ClassLoader cl : classLoaders) {
        classLoaderMap.put(getClassLoaderKey(cl), cl);
      }
    }

    @Override
    public ClassLoader create(String key) throws Exception {
      return duplicateClassLoader(classLoaderMap.get(key));
    }

    @Override
    public PooledObject<ClassLoader> wrap(ClassLoader value) {
      return new DefaultPooledObject<>(value);
    }
  }

  @Override
  public void initTask() {
    super.initTask();
    stageClassLoaders = runtimeInfo.getStageLibraryClassLoaders();
    if (!stageClassLoaders.isEmpty()) {
      resolveClassLoaderMethods(stageClassLoaders.get(0));
    }

    if(configuration.get(CONFIG_CP_VALIDATION, DEFAULT_CP_VALIDATION)) {
      validateStageClasspaths();
    }

    // Load all stages and other objects from the libraries
    json = ObjectMapperFactory.get();
    stageLibraries = new ArrayList<>();
    stageList = new ArrayList<>();
    stageMap = new HashMap<>();
    lineagePublisherDefinitions = new ArrayList<>();
    lineagePublisherDefinitionMap = new HashMap<>();
    credentialStoreDefinitions = new ArrayList<>();
    serviceList = new ArrayList<>();
    serviceMap = new HashMap<>();
    interceptorList = new ArrayList<>();
    delegateList = new ArrayList<>();
    delegateMap = new HashMap<>();
    loadStages();
    stageLibraries = ImmutableList.copyOf(stageLibraries);
    stageList = ImmutableList.copyOf(stageList);
    stageMap = ImmutableMap.copyOf(stageMap);
    lineagePublisherDefinitions = ImmutableList.copyOf(lineagePublisherDefinitions);
    lineagePublisherDefinitionMap = ImmutableMap.copyOf(lineagePublisherDefinitionMap);
    credentialStoreDefinitions = ImmutableList.copyOf(credentialStoreDefinitions);
    serviceList = ImmutableList.copyOf(serviceList);
    serviceMap = ImmutableMap.copyOf(serviceMap);
    interceptorList = ImmutableList.copyOf(interceptorList);
    delegateList = ImmutableList.copyOf(delegateList);
    delegateMap = ImmutableMap.copyOf(delegateMap);

    // localization cache for definitions
    localizedStageList = CacheBuilder.newBuilder().build(new CacheLoader<Locale, List<StageDefinition>>() {
      @Override
      public List<StageDefinition> load(Locale key) throws Exception {
        List<StageDefinition> list = new ArrayList<>();
        for (StageDefinition stage : stageList) {
          list.add(stage.localize());
        }
        return list;
      }
    });
    validateAllServicesAvailable();
    validateStageVersions(stageList);
    validateServices(stageList, serviceList);
    validateDelegates(delegateList);
    validateRequiredStageLibraries();

    // initializing the list of targets that can be used for error handling
    ErrorHandlingChooserValues.setErrorHandlingOptions(this);

    // initializing the list of targets that can be used as aggregating sink
    StatsTargetChooserValues.setStatsTargetOptions(this);

    // initializing the list of targets that can be used for pipeline lifecycle events
    PipelineLifecycleStageChooserValues.setHandlingOptions(this);

    // initializing the list of sources that can be used for test stages
    PipelineTestStageChooserValues.setHandlingOptions(this);

    // initializing the pool of private stage classloaders
    GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig();
    poolConfig.setJmxEnabled(false);
    poolConfig.setMaxTotal(configuration.get(MAX_PRIVATE_STAGE_CLASS_LOADERS_KEY,
                                             MAX_PRIVATE_STAGE_CLASS_LOADERS_DEFAULT));
    poolConfig.setMinEvictableIdleTimeMillis(-1);
    poolConfig.setNumTestsPerEvictionRun(0);
    poolConfig.setMaxIdlePerKey(-1);
    poolConfig.setMinIdlePerKey(0);
    poolConfig.setMaxTotalPerKey(-1);
    poolConfig.setBlockWhenExhausted(false);
    poolConfig.setMaxWaitMillis(0);
    privateClassLoaderPool = new GenericKeyedObjectPool<>(new ClassLoaderFactory(stageClassLoaders), poolConfig);
  }

  /**
   * Validate that all required libraries are available and loaded.
   */
  @VisibleForTesting void validateRequiredStageLibraries() {
    // Name of all installed libraries
    Set<String> installedLibraries = stageLibraries.stream()
      .map(StageLibraryDefinition::getName)
      .collect(Collectors.toSet());

    // Required libraries
    Set<String> requiredLibraries = new HashSet<>();
    String config = configuration.get(SdcConfiguration.REQUIRED_STAGELIBS, DEFAULT_REQUIRED_STAGELIBS);
    for(String stageLib : config.split(",")) {
      if(!stageLib.isEmpty()) {
        requiredLibraries.add(stageLib);
      }
    }

    Set<String> missingLibraries = Sets.difference(requiredLibraries, installedLibraries);
    if(!missingLibraries.isEmpty()) {
      throw new RuntimeException(Utils.format(
        "Some required stage libraries are missing: {}",
        StringUtils.join(missingLibraries, ", ")
      ));
    }
  }

  private void validateStageClasspaths() {
    LOG.info("Validating classpath of all stages");

    // Firstly validate the stage classpaths for duplicate dependencies
    Set<String> corruptedClasspathStages = new HashSet<>();
    for(ClasspathValidatorResult result : validateStageLibClasspath()) {
      if (!result.isValid()) {
        result.logDetails();
        corruptedClasspathStages.add(result.getName());
      }
    }

    if (corruptedClasspathStages.isEmpty()) {
      LOG.info("Classpath of all stages passed validation");
    } else {
      LOG.error("The following stages have invalid classpath: {}", StringUtils.join(corruptedClasspathStages, ", "));

      boolean canTerminate = configuration.get(CONFIG_CP_VALIDATION_RESULT, DEFAULT_CP_VALIDATION_RESULT);
      if (canTerminate) {
        throw new RuntimeException("Invalid classpath detected for " + corruptedClasspathStages.size() + " stage libraries: " + StringUtils.join(corruptedClasspathStages, ", "));
      }
    }
  }

  /**
   * Validate service dependencies.
   *
   * Any error is considered fatal and RuntimeException() will be thrown that will terminate the SDC start up procedure.
   */
  private void validateAllServicesAvailable() {
    // Firstly validate that all stages have satisfied service dependencies
    List<String> missingServices = new LinkedList<>();

    for(StageDefinition stage : stageList) {
      for(ServiceDependencyDefinition service : stage.getServices()) {
        if(!serviceMap.containsKey(service.getService())) {
          missingServices.add(Utils.format("Stage {} is missing service {}", stage.getName(), service.getService().getName()));
        }
      }
    }
    if(!missingServices.isEmpty()) {
      throw new RuntimeException("Missing services: " + StringUtils.join(missingServices, ", "));
    }

    // Secondly ensure that all loaded services are compatible with what is supported by our runtime engine
    List<String> unsupportedServices = new LinkedList<>();
    for(ServiceDefinition serviceDefinition : serviceList) {
      if(!ServiceRuntime.supports(serviceDefinition.getProvides())) {
        unsupportedServices.add(serviceDefinition.getProvides().toString());
      }
    }
    if(!unsupportedServices.isEmpty()) {
      throw new RuntimeException("Unsupported services: " + StringUtils.join(unsupportedServices, ", "));
    }
  }

  @Override
  protected void stopTask() {
    privateClassLoaderPool.close();
    super.stopTask();
  }

  Properties loadClasspathWhitelist(ClassLoader cl) {
   try (InputStream is = cl.getResourceAsStream(PROPERTIES_CP_WHITELIST)) {
      if (is != null) {
        Properties props = new Properties();
        props.load(is);
        return props;
      }
   } catch (IOException e) {
     // Fail over to null, the resources simply doesn't exists or something - since it's optional file, it's fully
     // legal to get here.
   }

    return null;
  }

  String getPropertyFromLibraryProperties(ClassLoader cl, String property, String defaultValue) throws IOException {
   try (InputStream is = cl.getResourceAsStream(StageLibraryDefinitionExtractor.DATA_COLLECTOR_LIBRARY_PROPERTIES)) {
      if (is != null) {
        Properties props = new Properties();
        props.load(is);
        return props.getProperty(property, defaultValue);
      }
    }

    return null;
  }

  Set<String> loadIgnoreStagesList(StageLibraryDefinition libDef) throws IOException {
    Set<String> ignoreStages = new HashSet<>();

    String ignore = getPropertyFromLibraryProperties(libDef.getClassLoader(), IGNORE_STAGE_DEFINITIONS, "");
    if(!StringUtils.isEmpty(ignore)) {
      ignoreStages.addAll(Splitter.on(",").trimResults().splitToList(ignore));
    }

    return ignoreStages;
  }

  List<String> removeIgnoreStagesFromList(StageLibraryDefinition libDef, List<String> stages) throws IOException {
    List<String> list = new ArrayList<>();
    Set<String> ignoreStages = loadIgnoreStagesList(libDef);
    Iterator<String> iterator = stages.iterator();
    while (iterator.hasNext()) {
      String stage = iterator.next();
      if (ignoreStages.contains(stage)) {
        LOG.debug("Ignoring stage class '{}' from library '{}'", stage, libDef.getName());
      } else {
        list.add(stage);
      }
    }
    return list;
  }

  @VisibleForTesting
  @SuppressWarnings("unchecked")
  void loadStages() {
    String javaVersion = System.getProperty("java.version");
    Version sdcVersion = new Version(buildInfo.getVersion());

    if (LOG.isDebugEnabled()) {
      for (ClassLoader cl : stageClassLoaders) {
        LOG.debug("Found stage library '{}'", StageLibraryUtils.getLibraryName(cl));
      }
    }

    try {
      RuntimeEL.loadRuntimeConfiguration(runtimeInfo);
      DataCollectorServices.instance().put(JsonMapper.SERVICE_KEY, new JsonMapperImpl());
    } catch (IOException e) {
      throw new RuntimeException(
        Utils.format("Could not load runtime configuration, '{}'", e.toString()), e);
    }

    try {
      int libs = 0;
      int stages = 0;
      int lineagePublishers = 0;
      int credentialStores = 0;
      int services = 0;
      int interceptors = 0;
      int delegates = 0;
      long start = System.currentTimeMillis();
      LocaleInContext.set(Locale.getDefault());
      for (ClassLoader cl : stageClassLoaders) {
        try {
          // Before loading any stages, let's verify that given stage library is compatible with our current JVM version
          String unsupportedJvmVersion = getPropertyFromLibraryProperties(cl, JAVA_UNSUPPORTED_REGEXP, null);
          if(!StringUtils.isEmpty(unsupportedJvmVersion)) {
            if(javaVersion.matches(unsupportedJvmVersion)) {
              LOG.warn("Can't load stages from {} since they are not compatible with current JVM version", StageLibraryUtils.getLibraryName(cl));
              continue;
            } else {
              LOG.debug("Stage lib {} passed java compatibility test for '{}'", StageLibraryUtils.getLibraryName(cl), unsupportedJvmVersion);
            }
          }

          // And that this SDC is at least on requested version
          String minSdcVersion = getPropertyFromLibraryProperties(cl, MIN_SDC_VERSION, null);
          if(!StringUtils.isEmpty(minSdcVersion)) {
            if(!sdcVersion.isGreaterOrEqualTo(minSdcVersion)) {
              throw new IllegalArgumentException(
                  Utils.format("Can't load stage library '{}' as it requires at least SDC version {} whereas current version is {}",
                  StageLibraryUtils.getLibraryName(cl),
                  minSdcVersion,
                  buildInfo.getVersion()
                ));
            }
          }

          // Load stages from the stage library
          StageLibraryDefinition libDef = StageLibraryDefinitionExtractor.get().extract(cl);
          LOG.debug("Loading stages and plugins from library '{}'", libDef.getName());
          stageLibraries.add(libDef);
          libs++;

          // Load Stages
          for(Class klass : loadClassesFromResource(libDef, cl, STAGES_DEFINITION_RESOURCE)) {
            stages++;
            StageDefinition stage = StageDefinitionExtractor.get().extract(libDef, klass, Utils.formatL("Library='{}'", libDef.getName()));
            String key = createKey(libDef.getName(), stage.getName());
            LOG.debug("Loaded stage '{}'  version {}", key, stage.getVersion());
            stageList.add(stage);
            stageMap.put(key, stage);
          }

          // Load Lineage publishers
          for(Class klass : loadClassesFromResource(libDef, cl, LINEAGE_PUBLISHERS_DEFINITION_RESOURCE)) {
            lineagePublishers++;
            LineagePublisherDefinition lineage = LineagePublisherDefinitionExtractor.get().extract(libDef, klass);
            String key = createKey(libDef.getName(), lineage.getName());
            LOG.debug("Loaded lineage plugin '{}'", key);
            lineagePublisherDefinitions.add(lineage);
            lineagePublisherDefinitionMap.put(key, lineage);
          }

          // Load Credential stores
          for(Class klass : loadClassesFromResource(libDef, cl, CREDENTIAL_STORE_DEFINITION_RESOURCE)) {
            credentialStores++;
            CredentialStoreDefinition def = CredentialStoreDefinitionExtractor.get().extract(libDef, klass);
            String key = createKey(libDef.getName(), def.getName());
            LOG.debug("Loaded credential store '{}'", key);
            credentialStoreDefinitions.add(def);
          }

          // Load Services
          for(Class klass : loadClassesFromResource(libDef, cl, SERVICE_DEFINITION_RESOURCE)) {
            services++;
            ServiceDefinition def = ServiceDefinitionExtractor.get().extract(libDef, klass);
            LOG.debug("Loaded service for '{}'", def.getProvides().getCanonicalName());
            serviceList.add(def);
            serviceMap.put(def.getProvides(), def);
          }

          // Load Interceptors
          for(Class klass : loadClassesFromResource(libDef, cl, INTERCEPTOR_DEFINITION_RESOURCE)) {
            interceptors++;
            InterceptorDefinition def = InterceptorDefinitionExtractor.get().extract(libDef, klass);
            LOG.debug("Loaded interceptor '{}'", def.getKlass().getCanonicalName());
            interceptorList.add(def);
          }

          // Load Delegates
          for(Class klass : loadClassesFromResource(libDef, cl, DELEGATE_DEFINITION_RESOURCE)) {
            delegates++;
            StageLibraryDelegateDefinitition def = StageLibraryDelegateDefinitionExtractor.get().extract(libDef, klass);
            String key = createKey(libDef.getName(), def.getExportedInterface().getCanonicalName());
            LOG.debug("Loaded delegate '{}'", def.getKlass().getCanonicalName());
            delegateList.add(def);
            delegateMap.put(key, def);
          }
        } catch (IOException | ClassNotFoundException ex) {
          throw new RuntimeException(
              Utils.format("Could not load stages definition from '{}', {}", cl, ex.toString()), ex);
        }
      }
      LOG.info(
        "Loaded '{}' libraries with a total of '{}' stages, '{}' lineage publishers, '{}' services, '{}' interceptors, '{}' delegates and '{}' credentialStores in '{}ms'",
        libs,
        stages,
        lineagePublishers,
        services,
        interceptors,
        delegates,
        credentialStores,
        System.currentTimeMillis() - start
      );
    } finally {
      LocaleInContext.set(null);
    }
  }

  private <T> List<Class<? extends T>> loadClassesFromResource(
    StageLibraryDefinition libDef,
    ClassLoader cl,
    String resourceName
  ) throws IOException, ClassNotFoundException {
    Set<String> dedup = new HashSet<>();
    List<Class<? extends T>> list = new ArrayList<>();

    // Load all resource files with given name
    Enumeration<URL> resources = cl.getResources(resourceName);
    while (resources.hasMoreElements()) {
      URL url = resources.nextElement();
      try (InputStream is = url.openStream()) {
        List<String> plugins = json.readValue(is, List.class);
        plugins = removeIgnoreStagesFromList(libDef, plugins);
        for (String className : plugins) {
          if(dedup.contains(className)) {
            throw new IllegalStateException(Utils.format(
              "Library '{}' contains more than one definition for '{}'",
              libDef.getName(), className));
          }
          dedup.add(className);
          list.add((Class<? extends T>) cl.loadClass(className));
        }
      }
    }

    return list;
  }

  @VisibleForTesting
  void validateStageVersions(List<StageDefinition> stageList) {
    boolean err = false;
    Map<String, Set<Integer>> stageVersions = new HashMap<>();
    for (StageDefinition stage : stageList) {
      Set<Integer> versions = stageVersions.get(stage.getName());
      if (versions == null) {
        versions = new HashSet<>();
        stageVersions.put(stage.getName(), versions);
      }
      versions.add(stage.getVersion());
      err |= versions.size() > 1;
    }
    if (err) {
      List<String> errors = new ArrayList<>();
      for (Map.Entry<String, Set<Integer>> entry : stageVersions.entrySet()) {
        if (entry.getValue().size() > 1) {
          for (StageDefinition stage : stageList) {
            if (stage.getName().equals(entry.getKey())) {
              errors.add(Utils.format("Stage='{}' Version='{}' Library='{}'", stage.getName(), stage.getVersion(),
                stage.getLibrary()));
            }
          }
        }
      }
      LOG.error("There cannot be 2 different versions of the same stage: {}", errors);
      throw new RuntimeException(Utils.format("There cannot be 2 different versions of the same stage: {}", errors));
    }
  }

  @VisibleForTesting
  void validateServices(List<StageDefinition> stages, List<ServiceDefinition> services) {
    List<String> errors = new ArrayList<>();

    // Firstly validate that there are no duplicate providers for each service
    Set<Class> duplicates = new HashSet<>();
    for(ServiceDefinition def: services) {
      if(duplicates.contains(def.getProvides())) {
        errors.add(Utils.format("Service {} have multiple implementations.", def.getProvides()));
      }

      duplicates.add(def.getProvides());
    }

    // TBD: Validate that we have services for all stage dependencies

    if(!errors.isEmpty()) {
      throw new RuntimeException(Utils.format("Validate errors when loading services: {}", errors));
    }
  }

  @VisibleForTesting
  void validateDelegates(List<StageLibraryDelegateDefinitition> delegateList) {
    Set<String> errors = new HashSet<>();
    Map<String, Set<Class>> delegatesPerStageLib = new HashMap<>();

    // The current validation is only that each delegate interface can be exported max once from given stage library
    for(StageLibraryDelegateDefinitition def : delegateList) {
      String name = def.getLibraryDefinition().getName();

      Set<Class> declaredDelegates = delegatesPerStageLib.computeIfAbsent(name, (n) -> new HashSet<>());

      if(declaredDelegates.contains(def.getExportedInterface())) {
        errors.add(Utils.format(
          "Stage library '{}' exports delegate for '{}' more then once",
          name,
          def.getExportedInterface().getCanonicalName()
        ));
      }

      declaredDelegates.add(def.getExportedInterface());

      if(!StageLibraryDelegateRuntime.supports(def.getExportedInterface())) {
        errors.add(Utils.format(
          "Delegate interface {} is not supported by this runtime",
          def.getExportedInterface()
        ));
      }
    }

    if(!errors.isEmpty()) {
      throw new RuntimeException(Utils.format("Validate errors when loading delegates: {}", errors));
    }
  }

  private String createKey(String library, String name) {
    return library + ":" + name;
  }

  @Override
  public PipelineDefinition getPipeline() {
    return PipelineDefinition.getPipelineDef();
  }

  @Override
  public PipelineFragmentDefinition getPipelineFragment() {
    return PipelineFragmentDefinition.getPipelineFragmentDef();
  }

  @Override
  public PipelineRulesDefinition getPipelineRules() {
    return PipelineRulesDefinition.getPipelineRulesDef();
  }

  @Override
  public List<StageDefinition> getStages() {
    try {
      return (LocaleInContext.get() == null) ? stageList : localizedStageList.get(LocaleInContext.get());
    } catch (ExecutionException ex) {
      LOG.warn("Error loading locale '{}', {}", LocaleInContext.get(), ex.toString(), ex);
      return stageList;
    }
  }

  @Override
  public List<LineagePublisherDefinition> getLineagePublisherDefinitions() {
    return lineagePublisherDefinitions;
  }

  @Override
  public LineagePublisherDefinition getLineagePublisherDefinition(String library, String name) {
    return lineagePublisherDefinitionMap.get(createKey(library, name));
  }

  @Override
  public List<CredentialStoreDefinition> getCredentialStoreDefinitions() {
    return credentialStoreDefinitions;
  }

  @Override
  public List<ServiceDefinition> getServiceDefinitions() {
    return serviceList;
  }

  @Override
  public ServiceDefinition getServiceDefinition(Class serviceInterface, boolean forExecution) {
    ServiceDefinition serviceDefinition = serviceMap.get(serviceInterface);

    if(forExecution && serviceDefinition.isPrivateClassLoader()) {
      serviceDefinition = new ServiceDefinition(serviceDefinition, getStageClassLoader(serviceDefinition));
    }

    return serviceDefinition;
  }

  @Override
  public List<InterceptorDefinition> getInterceptorDefinitions() {
    return interceptorList;
  }

  @Override
  @SuppressWarnings("unchecked")
  public StageDefinition getStage(String library, String name, boolean forExecution) {
    StageDefinition def = stageMap.get(createKey(library, name));
    if (forExecution &&  def.isPrivateClassLoader()) {
      def = new StageDefinition(def, getStageClassLoader(def));
    }
    return def;
  }

  @Override
  public Map<String, String> getLibraryNameAliases() {
    return libraryNameAliases;
  }

  @Override
  public Map<String, String> getStageNameAliases() {
    return stageNameAliases;
  }

  @Override
  public List<ClasspathValidatorResult> validateStageLibClasspath() {
    long startTime = System.currentTimeMillis();
    List<ClasspathValidatorResult> validators = new LinkedList<>();

    for (ClassLoader cl : stageClassLoaders) {
      if (cl instanceof SDCClassLoader) {
        SDCClassLoader sdcCl = (SDCClassLoader) cl;

        ClasspathValidatorResult validationResult = ClasspathValidator.newValidator(sdcCl.getName())
          .withURLs(sdcCl.getURLs())
          .validate(loadClasspathWhitelist(cl));

        validators.add(validationResult);
      }
    }

    LOG.info("Finished classpath validation in {} ms", System.currentTimeMillis() - startTime);
    return validators;
  }

  @Override
  public List<StageLibraryDelegateDefinitition> getStageLibraryDelegateDefinitions() {
    return delegateList;
  }

  @Override
  public StageLibraryDelegateDefinitition getStageLibraryDelegateDefinition(String stageLibrary, Class exportedInterface) {
    return delegateMap.get(createKey(stageLibrary, exportedInterface.getCanonicalName()));
  }

  @Override
  public List<StageLibraryDefinition> getLoadedStageLibraries() {
    return stageLibraries;
  }

  ClassLoader getStageClassLoader(PrivateClassLoaderDefinition stageDefinition) {
    ClassLoader cl = stageDefinition.getStageClassLoader();
    if (stageDefinition.isPrivateClassLoader()) {
      String key = getClassLoaderKey(cl);
      synchronized (privateClassLoaderPool) {
        try {
          cl = privateClassLoaderPool.borrowObject(key);
          LOG.debug("Got a private ClassLoader for '{}', for '{}', active private ClassLoaders='{}'",
              key, stageDefinition.getName(), privateClassLoaderPool.getNumActive());
        } catch (Exception ex) {
          String msg = Utils.format(
              "Could not get a private ClassLoader for '{}', for '{}', active private ClassLoaders='{}': {}",
              key, stageDefinition.getName(), privateClassLoaderPool.getNumActive(), ex.toString());
          LOG.warn(msg, ex);
          throw new RuntimeException(msg, ex);
        }
      }
    }
    return cl;
  }

  @Override
  public void releaseStageClassLoader(ClassLoader classLoader) {
    if (isPrivateClassLoader(classLoader)) {
      String key = getClassLoaderKey(classLoader);
      synchronized (privateClassLoaderPool){
        if (privateClassLoaderPool.getNumActive() > 0) {
          try {
            LOG.debug("Returning private ClassLoader for '{}'", key);
            privateClassLoaderPool.returnObject(key, classLoader);
            LOG.debug("Returned a private ClassLoader for '{}', active private ClassLoaders='{}'",
                key, privateClassLoaderPool.getNumActive());
          } catch (Exception ex) {
            LOG.warn("Could not return a private ClassLoader for '{}', active private ClassLoaders='{}'",
                key, privateClassLoaderPool.getNumActive());
            throw new RuntimeException(ex);
          }
        }
      }
    }
  }

  @Override
  public List<RepositoryManifestJson> getRepositoryManifestList() {
    if (repositoryManifestList == null) {
      // initialize when it is called for first time
      repositoryManifestList = new ArrayList<>();

      // Initialize Repository Links
      String [] repoURLList;
      String repoLinksStr = configuration.get(CONFIG_PACKAGE_MANAGER_REPOSITORY_LINKS, "");
      if (StringUtils.isEmpty(repoLinksStr)) {
        String version = buildInfo.getVersion();
        String repoUrl = ARCHIVES_URL + version + TARBALL_PATH;
        if (version.contains(SNAPSHOT)) {
          repoUrl = NIGHTLY_URL + LATEST + TARBALL_PATH;
        }
        repoURLList = new String[] {
            repoUrl,
            repoUrl + ENTERPRISE_PATH
        };
      } else {
        repoURLList = repoLinksStr.split(",");
      }

      List<StageLibraryManifestJson> installedLibraries = new ArrayList<>();
      List<StageLibraryManifestJson> additionalLibraries = new ArrayList<>();

      Map<String, List<StageInfoJson>> installedStagesMap = new HashMap<>();
      for(StageDefinition stageDefinition: getStages()) {
        List<StageInfoJson> stagesList;
        if (installedStagesMap.containsKey(stageDefinition.getLibrary())) {
          stagesList = installedStagesMap.get(stageDefinition.getLibrary());
        } else {
          stagesList = new ArrayList<>();
          installedStagesMap.put(stageDefinition.getLibrary(), stagesList);
        }
        stagesList.add(new StageInfoJson(stageDefinition));
      }

      Map<String, Boolean> installedLibrariesMap = new HashMap<>();
      for(StageLibraryDefinition libDef : getLoadedStageLibraries()) {
        installedLibrariesMap.put(libDef.getName(), true);
        installedLibraries.add(new StageLibraryManifestJson(
            libDef.getName(),
            libDef.getLabel(),
            installedStagesMap.getOrDefault(libDef.getName(), Collections.emptyList()),
            true
        ));
      }

      Set<String> addedLibraryIds = new HashSet<>();

      for (String repoUrl: repoURLList) {
        if (!repoUrl.endsWith("/")) {
          repoUrl = repoUrl + "/";
        }
        String repoManifestUrl = repoUrl +  REPOSITORY_MANIFEST_JSON_PATH;
        LOG.info("Reading from Repository Manifest URL: " + repoManifestUrl);
        RepositoryManifestJson repositoryManifestJson = getRepositoryManifestFile(repoManifestUrl);
        if (repositoryManifestJson != null) {
          repositoryManifestJson.setRepoUrl(repoUrl);
          for(StageLibrariesJson stageLibrariesJson: repositoryManifestJson.getStageLibraries()) {
            String stageLibManifestUrl = repoUrl + stageLibrariesJson.getStagelibManifest();
            StageLibraryManifestJson stageLibraryManifestJson = getStageLibraryManifestJson(stageLibManifestUrl);
            if (stageLibraryManifestJson != null) {
              stageLibraryManifestJson.setInstalled(
                  installedLibrariesMap.containsKey(stageLibraryManifestJson.getStageLibId())
              );
              stageLibraryManifestJson.setStageLibFile(repoUrl + stageLibraryManifestJson.getStageLibFile());
              stageLibrariesJson.setStageLibraryManifest(stageLibraryManifestJson);
              addedLibraryIds.add(stageLibraryManifestJson.getStageLibId());
            }
          }
          repositoryManifestList.add(repositoryManifestJson);
        }
      }

      // Add installed custom/user stage libraries to the list which are not part of Archives manifest file
      List<StageLibrariesJson> additionalList = new ArrayList<>();
      for (StageLibraryManifestJson installedLibrary : installedLibraries) {
        if (!addedLibraryIds.contains(installedLibrary.getStageLibId())) {
          additionalLibraries.add(installedLibrary);
          StageLibrariesJson stageLibrariesJson = new StageLibrariesJson();
          stageLibrariesJson.setStageLibraryManifest(installedLibrary);
          additionalList.add(stageLibrariesJson);
        }
      }

      if (!additionalLibraries.isEmpty()) {
        RepositoryManifestJson additionalRepo = new RepositoryManifestJson();
        additionalRepo.setRepoUrl(ADDITIONAL);
        additionalRepo.setRepoLabel(ADDITIONAL);
        additionalRepo.setStageLibraries(additionalList);
        repositoryManifestList.add(additionalRepo);
      }
    }

    return repositoryManifestList;
  }

  @Override
  public boolean isMultipleOriginSupported() {
    return false;
  }

  private RepositoryManifestJson getRepositoryManifestFile(String repoUrl) {
    RepositoryManifestJson repositoryManifestJson = null;
    try (Response response = ClientBuilder.newClient().target(repoUrl).request().get()) {
      InputStream inputStream = response.readEntity(InputStream.class);
      repositoryManifestJson = ObjectMapperFactory.get().readValue(inputStream, RepositoryManifestJson.class);
    } catch (Exception ex) {
      LOG.error("Failed to read repository manifest json", ex);
    }
    return repositoryManifestJson;
  }

  private StageLibraryManifestJson getStageLibraryManifestJson(String stageLibManifestUrl) {
    StageLibraryManifestJson stageLibManifestJson = null;
    try (Response response = ClientBuilder.newClient().target(stageLibManifestUrl).request().get()) {
      InputStream inputStream = response.readEntity(InputStream.class);
      stageLibManifestJson = ObjectMapperFactory.get().readValue(inputStream, StageLibraryManifestJson.class);
    }  catch (Exception ex) {
      LOG.error("Failed to read stage-lib-manifest.json", ex);
    }
    return stageLibManifestJson;
  }

}


text:
<?xml version="1.0" encoding="utf-8"?>
<selector xmlns:android="http://schemas.android.com/apk/res/android">
    <item android:state_enabled="true" android:drawable="@drawable/auto_traffic_rightarrow" />
    <item android:state_enabled="false" android:drawable="@drawable/auto_traffic_rightarrow_unable" />
</selector>

redpajama_set_name: RedPajamaGithub


text:
module CmeFixListener
  # <TrdRegTS> Mappings from abbreviations to full names (with types)
  class SideTradeRegulatoryParser
    extend ParsingMethods

    MAPPINGS = [
      ["timestamp", "TS"],
      ["timestampType", "Typ", :to_i]
    ].freeze
  end
end


text:
var machine = require("./machine");
var assert = require("assert");

// I'd normally use mocha but come on, that's just an exercise - why add dependencies?
function testSuite(withdraw) {
    // successful scenarios
    assert.deepStrictEqual(withdraw(0), [0, 0, 0, 0]);
    assert.deepStrictEqual(withdraw(10), [0, 0, 0, 1]);
    assert.deepStrictEqual(withdraw(20), [0, 0, 1, 0]);
    assert.deepStrictEqual(withdraw(30), [0, 0, 1, 1]);
    assert.deepStrictEqual(withdraw(40), [0, 0, 2, 0]);
    assert.deepStrictEqual(withdraw(50), [0, 1, 0, 0]);
    assert.deepStrictEqual(withdraw(60), [0, 1, 0, 1]);
    assert.deepStrictEqual(withdraw(70), [0, 1, 1, 0]);
    assert.deepStrictEqual(withdraw(80), [0, 1, 1, 1]);
    assert.deepStrictEqual(withdraw(90), [0, 1, 2, 0]);
    assert.deepStrictEqual(withdraw(100), [1, 0, 0, 0]);
    assert.deepStrictEqual(withdraw(500), [5, 0, 0, 0]);
    assert.deepStrictEqual(withdraw(780), [7, 1, 1, 1]);
    assert.deepStrictEqual(withdraw(11110), [111, 0, 0, 1]);

    // failures
    try {
        withdraw(55);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "Cannot deliver 55 in available notes. Withdrawal amount must end with a zero (10, 80, ...)")
    }

    try {
        withdraw(8521);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "Cannot deliver 8521 in available notes. Withdrawal amount must end with a zero (10, 80, ...)")
    }

    try {
        withdraw();
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "undefined is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw(null);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "null is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw("4.50");
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "4.50 is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw("420");
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "420 is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw(Number.NaN);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "NaN is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw(Infinity);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "Infinity is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }

    try {
        withdraw(4.50);
        assert.fail("Should not be able to withdraw that");
    } catch (err) {
        assert.strictEqual(err.msg, "4.5 is not deliverable. Withdrawal amount must be a positive integer ending with 0 (10, 80, ...)")
    }
}

testSuite(machine.withdraw);
testSuite(machine.fastWithdraw);


text:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;

namespace Obvs.MessageDispatcher
{
    internal sealed class DefaultMessageHandlerSelector : IDefaultMessageHandlerSelector
    {
        internal Dictionary<Type, Func<IMessageHandler>> _messageHandlerTypesByHandledMessageType = new Dictionary<Type, Func<IMessageHandler>>();

        public DefaultMessageHandlerSelector()
        {
        }

        public void RegisterMessageHandler(Type messageHandlerType)
        {
            ForEachSupportedMessageType(messageHandlerType, supportedMessageType => _messageHandlerTypesByHandledMessageType[supportedMessageType] = Expression.Lambda<Func<IMessageHandler>>(Expression.New(messageHandlerType)).Compile());
        }

        public void RegisterMessageHandler(IMessageHandler messageHandler)
        {
            if(messageHandler == null) throw new ArgumentNullException(nameof(messageHandler));

            ForEachSupportedMessageType(messageHandler.GetType(), supportedMessageType => _messageHandlerTypesByHandledMessageType[supportedMessageType] = () => messageHandler);
        }

        public void RegisterMessageHandler<TMessageHandler>(Func<TMessageHandler> messageHandlerFactory)
            where TMessageHandler : class
        {
            if(messageHandlerFactory == null) throw new ArgumentNullException(nameof(messageHandlerFactory));

            ForEachSupportedMessageType(typeof(TMessageHandler), supportedMessageType => _messageHandlerTypesByHandledMessageType[supportedMessageType] = (Func<IMessageHandler>)messageHandlerFactory);
        }

        public IMessageHandler<TMessage> SelectMessageHandler<TMessage>(TMessage message)
        {
            Func<IMessageHandler> messageHandlerFactoryForMessageType;

            _messageHandlerTypesByHandledMessageType.TryGetValue(typeof(TMessage), out messageHandlerFactoryForMessageType);

            return (IMessageHandler<TMessage>)messageHandlerFactoryForMessageType();
        }

        private void ForEachSupportedMessageType(Type messageHandlerType, Action<Type> action)
        {
            if(messageHandlerType == null) throw new ArgumentNullException(nameof(messageHandlerType));
            if(messageHandlerType.GetConstructor(Type.EmptyTypes) == null) throw new ArgumentException($"The specified message handler type, ${messageHandlerType.Name}, does not contain an empty constructor.", nameof(messageHandlerType));

            foreach(Type messageType in messageHandlerType.GetInterfaces().Where(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IMessageHandler<>)).SelectMany(i => i.GetGenericArguments()))
            {
                action(messageType);
            }
        }
    }
}


text:
#include "tensorflow/compiler/xla/pjrt/gpu_device.h"

#include "absl/base/attributes.h"
#include "absl/container/flat_hash_map.h"
#include "tensorflow/compiler/xla/pjrt/pjrt_stream_executor_client.h"
#include "tensorflow/stream_executor/device_memory.h"

#ifdef GOOGLE_CUDA
#include "third_party/gpus/cuda/include/cuda.h"
#include "tensorflow/stream_executor/cuda/cuda_activation.h"
#endif  // GOOGLE_CUDA

#ifdef NCCL_ENABLED
#include "third_party/nccl/nccl.h"
#endif  // NCCL_ENABLED
#include "tensorflow/compiler/xla/client/client_library.h"
#include "tensorflow/compiler/xla/service/gpu/gpu_executable_run_options.h"
#include "tensorflow/compiler/xla/service/platform_util.h"
#include "tensorflow/compiler/xla/statusor.h"
#include "tensorflow/compiler/xla/util.h"
#include "tensorflow/core/common_runtime/device/device_host_allocator.h"
#include "tensorflow/core/common_runtime/device/device_id.h"
#include "tensorflow/core/common_runtime/device/device_mem_allocator.h"
#include "tensorflow/core/util/env_var.h"
#include "tensorflow/stream_executor/tf_allocator_adapter.h"

namespace xla {
namespace {

#if defined(GOOGLE_CUDA) && CUDA_VERSION >= 11020

std::string GetCudaErrorMessage(CUresult result) {
  const char* error;
  cuGetErrorString(result, &error);
  const char* name;
  cuGetErrorName(result, &name);
  return absl::StrCat("CUDA error: ", error ? error : "<unknown>", " (",
                      name ? name : "Unknown", ")");
}

// A compute-stream synchronized allocator, implemented using the CUDA async
// allocation API added in CUDA 11.2.
// TODO(phawkins): this approach does not use the full capabilities of the
// allocator. We don't need to synchronize allocations to the compute stream
// with this allocator design. However that would be a larger change to PJRT.
class CudaAsyncDeviceMemoryAllocator : public se::DeviceMemoryAllocator {
 public:
  static StatusOr<std::unique_ptr<CudaAsyncDeviceMemoryAllocator>> Create(
      se::Platform* platform, std::vector<se::Stream*> streams) {
    auto allocator = std::make_unique<CudaAsyncDeviceMemoryAllocator>(platform);
    allocator->pools_.resize(streams.size());

    for (size_t i = 0; i < streams.size(); ++i) {
      se::Stream* stream = streams[i];
      TF_RET_CHECK(stream->parent()->device_ordinal() == i);
      se::cuda::ScopedActivateExecutorContext scoped_activation{
          stream->parent()};
      int cuda_malloc_async_supported;
      if (auto status = cuDeviceGetAttribute(
              &cuda_malloc_async_supported,
              CU_DEVICE_ATTRIBUTE_MEMORY_POOLS_SUPPORTED, i)) {
        return Unknown("Failed to get device attribute: %s",
                       GetCudaErrorMessage(status));
      }
      if (!cuda_malloc_async_supported) {
        return FailedPrecondition(
            "cuda_malloc_async isn't supported. "
            " Possible causes: device not supported, driver too old, "
            " OS not supported, CUDA version too old.");
      }
      if (auto status = cuDeviceGetDefaultMemPool(&allocator->pools_[i], i)) {
        return Unknown("Failed to get default CUDA pool: %s",
                       GetCudaErrorMessage(status));
      }
    }
    allocator->streams_ = std::move(streams);
    return allocator;
  }

  // Use Create() instead of calling this constructor.
  explicit CudaAsyncDeviceMemoryAllocator(se::Platform* platform)
      : se::DeviceMemoryAllocator(platform) {}

  StatusOr<se::OwningDeviceMemory> Allocate(int device_ordinal, uint64_t size,
                                            bool retry_on_failure,
                                            int64_t memory_space) override {
    se::Stream* stream = streams_.at(device_ordinal);
    se::cuda::ScopedActivateExecutorContext scoped_activation{stream->parent()};
    CUstream custream = reinterpret_cast<cudaStream_t>(
        stream->implementation()->GpuStreamHack());
    void* ptr = nullptr;
    if (auto result =
            cuMemAllocFromPoolAsync(reinterpret_cast<CUdeviceptr*>(&ptr), size,
                                    pools_.at(device_ordinal), custream)) {
      return ResourceExhausted("CUDA allocation of %d bytes failed.", size);
    }
    return se::OwningDeviceMemory(se::DeviceMemoryBase(ptr, size),
                                  device_ordinal, this);
  }

  Status Deallocate(int device_ordinal, se::DeviceMemoryBase mem) override {
    se::Stream* stream = streams_.at(device_ordinal);
    se::cuda::ScopedActivateExecutorContext scoped_activation{stream->parent()};
    CUstream custream = reinterpret_cast<cudaStream_t>(
        stream->implementation()->GpuStreamHack());
    void* ptr = const_cast<void*>(mem.opaque());
    if (auto result = cuMemFreeAsync(reinterpret_cast<const CUdeviceptr&>(ptr),
                                     custream)) {
      return Unknown("CUDA deallocation failed.");
    }
    return Status::OK();
  }

  StatusOr<se::Stream*> GetStream(int device_ordinal) override {
    return streams_.at(device_ordinal);
  }

 private:
  std::vector<se::Stream*> streams_;
  std::vector<CUmemoryPool> pools_;
};

StatusOr<std::unique_ptr<se::DeviceMemoryAllocator>> CreateCudaAsyncAllocator(
    se::Platform* platform,
    absl::Span<std::unique_ptr<LocalDeviceState> const> addressable_devices) {
  std::vector<se::Stream*> streams;
  streams.reserve(addressable_devices.size());
  for (const auto& device : addressable_devices) {
    streams.push_back(device->compute_stream());
  }
  return CudaAsyncDeviceMemoryAllocator::Create(platform, std::move(streams));
}

#else  // defined(GOOGLE_CUDA) && CUDA_VERSION >= 11020

StatusOr<std::unique_ptr<se::DeviceMemoryAllocator>> CreateCudaAsyncAllocator(
    se::Platform* platform,
    absl::Span<std::unique_ptr<LocalDeviceState> const> addressable_devices) {
  return FailedPrecondition("CUDA async allocator requires CUDA >= 11.2");
}

#endif  // defined(GOOGLE_CUDA) && CUDA_VERSION >= 11020

// A custom PjRtClient that overrides the device assignment method.
class GpuClient : public xla::PjRtStreamExecutorClient {
 public:
  using xla::PjRtStreamExecutorClient::PjRtStreamExecutorClient;

  xla::StatusOr<xla::DeviceAssignment> GetDefaultDeviceAssignment(
      int num_replicas, int num_partitions) const override;
};

xla::StatusOr<xla::DeviceAssignment> GpuClient::GetDefaultDeviceAssignment(
    int num_replicas, int num_partitions) const {
  if (num_partitions == 1 && num_replicas <= addressable_devices().size()) {
    xla::DeviceAssignment assignment(num_replicas, 1);
    for (int i = 0; i < num_replicas; ++i) {
      assignment(i, 0) = addressable_devices().at(i)->id();
    }
    return assignment;
  }
  // Fallback to default global device assignment if we can't run locally.
  return PjRtStreamExecutorClient::GetDefaultDeviceAssignment(num_replicas,
                                                              num_partitions);
}

// Builds an xla::LocalClient for the GPU platform.
StatusOr<LocalClient*> GetGpuXlaClient() {
  // "gpu" will be substitued by the default defined in platform_util.cc
  TF_ASSIGN_OR_RETURN(se::Platform * platform,
                      PlatformUtil::GetPlatform("gpu"));
  if (platform->VisibleDeviceCount() <= 0) {
    return FailedPrecondition("No visible GPU devices.");
  }
  LocalClientOptions options;
  options.set_platform(platform);
  return ClientLibrary::GetOrCreateLocalClient(options);
}

// Builds a LocalDeviceState for each GPU present.
StatusOr<std::vector<std::unique_ptr<LocalDeviceState>>> BuildLocalDeviceStates(
    LocalClient* xla_client, bool asynchronous) {
  std::vector<std::unique_ptr<LocalDeviceState>> addressable_devices;
  for (int i = 0; i < xla_client->device_count(); ++i) {
    se::StreamExecutor* executor =
        xla_client->backend().stream_executor(i).ValueOrDie();
    addressable_devices.push_back(absl::make_unique<LocalDeviceState>(
        executor, xla_client, LocalDeviceState::kComputeSynchronized,
        asynchronous,
        /*allow_event_reuse=*/true));
  }
  return std::move(addressable_devices);
}

// Builds a BFCAllocator for all local GPUs.
StatusOr<std::unique_ptr<se::MultiDeviceAdapter>> CreateBFCAllocator(
    absl::Span<std::unique_ptr<LocalDeviceState> const> addressable_devices,
    double memory_fraction, bool preallocate) {
  CHECK_GT(addressable_devices.size(), 0);
  const se::Platform* platform =
      addressable_devices.front()->executor()->platform();
  std::vector<se::MultiDeviceAdapter::AllocatorWithStream> allocators;
  bool enable_unified_memory;
  Status status = tensorflow::ReadBoolFromEnvVar("TF_FORCE_UNIFIED_MEMORY",
                                                 false, &enable_unified_memory);
  if (!status.ok()) {
    LOG(ERROR) << "Unable to read TF_FORCE_UNIFIED_MEMORY: "
               << status.error_message();
  }

  for (auto& local_device : addressable_devices) {
    se::StreamExecutor* executor = local_device->executor();
    int device_ordinal = executor->device_ordinal();
    auto sub_allocator = absl::make_unique<tensorflow::DeviceMemAllocator>(
        executor, tensorflow::PlatformDeviceId(device_ordinal),
        /*use_unified_memory=*/enable_unified_memory,
        /*alloc_visitors=*/std::vector<tensorflow::SubAllocator::Visitor>(),
        /*free_visitors=*/std::vector<tensorflow::SubAllocator::Visitor>());

    int64 free_memory;
    int64 total_memory;
    if (!executor->DeviceMemoryUsage(&free_memory, &total_memory)) {
      return Unavailable("Failed to query available memory from device %i",
                         device_ordinal);
    }
    // To allow full GPU memory to be visible to the BFC allocator if using
    // unified memory.
    size_t allocator_memory =
        enable_unified_memory ? total_memory : free_memory * memory_fraction;
    if (preallocate) {
      LOG(INFO) << "XLA backend allocating " << allocator_memory
                << " bytes on device " << device_ordinal
                << " for BFCAllocator.";
    } else {
      LOG(INFO) << "XLA backend will use up to " << allocator_memory
                << " bytes on device " << device_ordinal
                << " for BFCAllocator.";
    }
    auto gpu_bfc_allocator = absl::make_unique<tensorflow::BFCAllocator>(
        sub_allocator.release(), allocator_memory,
        /*allow_growth=*/!preallocate,
        absl::StrCat("GPU_", device_ordinal, "_bfc"));
    allocators.emplace_back(std::move(gpu_bfc_allocator),
                            local_device->compute_stream());
  }
  return absl::make_unique<se::MultiDeviceAdapter>(platform,
                                                   std::move(allocators));
}

// Constructs a GPU device memory allocator to use, according to the allocator
// configuration the client requested.
StatusOr<std::unique_ptr<se::DeviceMemoryAllocator>> GetGpuDeviceAllocator(
    se::Platform* platform, const GpuAllocatorConfig& allocator_config,
    absl::Span<std::unique_ptr<LocalDeviceState> const> addressable_devices) {
  std::unique_ptr<se::DeviceMemoryAllocator> allocator;
  switch (allocator_config.kind) {
    case GpuAllocatorConfig::Kind::kCudaAsync: {
      auto allocator_or =
          CreateCudaAsyncAllocator(platform, addressable_devices);
      if (allocator_or.ok()) {
        LOG(INFO) << "Using CUDA async allocator.";
        allocator = std::move(allocator_or.ValueOrDie());
        break;
      }
      LOG(ERROR) << "Failed to initialize CUDA async allocator: "
                 << allocator_or.status() << "; falling back to BFC.";
      ABSL_FALLTHROUGH_INTENDED;
    }

    case GpuAllocatorConfig::Kind::kDefault:
    case GpuAllocatorConfig::Kind::kBFC: {
      LOG(INFO) << "Using BFC allocator.";
      TF_ASSIGN_OR_RETURN(allocator,
                          CreateBFCAllocator(addressable_devices,
                                             allocator_config.memory_fraction,
                                             allocator_config.preallocate));
      break;
    }

    case GpuAllocatorConfig::Kind::kPlatform:
      LOG(INFO) << "Using platform allocator.";
      break;
  }
  return std::move(allocator);
}

// Returns a GPU pinned host memory allocator to use when staging host->GPU
// transfers. We use a fixed 64MB pool of pinned memory.
std::unique_ptr<tensorflow::BFCAllocator> GetGpuHostAllocator(
    se::StreamExecutor* executor) {
  tensorflow::SubAllocator* sub_allocator = new tensorflow::DeviceHostAllocator(
      executor, /*numa_node=*/0, /*alloc_visitors=*/{}, /*free_visitors=*/{});
  // TODO(phawkins): allow the user to tune this.
  const int64 kGpuHostMemoryLimitBytes = 64 * (1LL << 30);
  return absl::make_unique<tensorflow::BFCAllocator>(
      sub_allocator, kGpuHostMemoryLimitBytes, /*allow_growth=*/true,
      /*name=*/"xla_gpu_host_bfc");
}

// A table mapping NcclCliqueKeys to ncclUniqueId values encoded as strings.
// In a distributed setup the table of NCCL IDs is kept on the master node
// (node 0). The node of the first participating device will create the unique
// id.
class NcclIdStore {
 public:
  NcclIdStore(int node_id, std::shared_ptr<DistributedRuntimeClient> client,
              absl::flat_hash_map<GlobalDeviceId, int> device_to_node)
      : node_id_(node_id),
        client_(std::move(client)),
        device_to_node_(std::move(device_to_node)) {}

  StatusOr<std::string> GetNcclUniqueId(const gpu::NcclCliqueKey& key);

 private:
  const int node_id_;
  const std::shared_ptr<DistributedRuntimeClient> client_;
  const absl::flat_hash_map<GlobalDeviceId, int> device_to_node_;

  absl::Mutex mu_;
  absl::flat_hash_map<gpu::NcclCliqueKey, std::string> cache_
      ABSL_GUARDED_BY(mu_);
};

StatusOr<std::string> NcclIdStore::GetNcclUniqueId(
    const gpu::NcclCliqueKey& key) {
  // The caller must ensure that threads calling this method concurrently have
  // unique keys, otherwise the global key-value store may hold the wrong value.
  {
    absl::MutexLock lock(&mu_);
    auto it = cache_.find(key);
    if (it != cache_.end()) {
      return it->second;
    }
  }
  std::string id_string;
  int primary_node_id = device_to_node_.at(key.devices()[0]);
  if (node_id_ == primary_node_id) {
#ifdef NCCL_ENABLED
    ncclUniqueId id;
    ncclResult_t r = ncclGetUniqueId(&id);
    TF_RET_CHECK(r == ncclSuccess);
    id_string = std::string(id.internal, NCCL_UNIQUE_ID_BYTES);
    TF_RETURN_IF_ERROR(client_->KeyValueSet(key.ToString(), id_string));
#else
    return FailedPrecondition("NCCL support was not built into XLA binary.");
#endif
  } else {
    TF_ASSIGN_OR_RETURN(id_string, client_->BlockingKeyValueGet(
                                       key.ToString(), absl::Minutes(5)));
  }
  absl::MutexLock lock(&mu_);
  auto result = cache_.emplace(key, std::move(id_string));
  TF_RET_CHECK(result.second) << "Unique ID already in cache.";
  return result.first->second;
}

std::vector<std::unique_ptr<PjRtStreamExecutorDevice>> BuildLocalDevices(
    std::vector<std::unique_ptr<LocalDeviceState>> local_device_states) {
  std::vector<std::unique_ptr<PjRtStreamExecutorDevice>> devices;
  for (auto& local_device : local_device_states) {
    int device_ordinal = local_device->device_ordinal();
    const se::DeviceDescription& description =
        local_device->executor()->GetDeviceDescription();
    auto device = absl::make_unique<GpuDevice>(
        device_ordinal, std::move(local_device), description.name(),
        /*node_id=*/0);
    devices.push_back(std::move(device));
  }
  return devices;
}

Status BuildDistributedDevices(
    std::vector<std::unique_ptr<LocalDeviceState>> local_device_states,
    std::shared_ptr<DistributedRuntimeClient> distributed_client, int node_id,
    std::vector<std::unique_ptr<PjRtStreamExecutorDevice>>* devices,
    gpu::GpuExecutableRunOptions* gpu_executable_run_options) {
  LocalTopologyProto local_topology;
  local_topology.set_node_id(node_id);
  for (const auto& local_device : local_device_states) {
    const se::Platform* platform = local_device->executor()->platform();
    TF_ASSIGN_OR_RETURN(
        std::unique_ptr<xla::se::DeviceDescription> desc,
        platform->DescriptionForDevice(local_device->device_ordinal()));
    TF_RET_CHECK(local_device->device_ordinal() ==
                 local_topology.devices_size());
    DeviceProto* device_proto = local_topology.add_devices();
    device_proto->set_local_device_ordinal(local_device->device_ordinal());
    device_proto->set_name(desc->name());
    device_proto->set_vendor(desc->device_vendor());
  }

  GlobalTopologyProto global_topology;
  TF_RETURN_IF_ERROR(
      distributed_client->EnumerateDevices(local_topology, &global_topology));

  std::vector<GlobalDeviceId> gpu_device_ids(local_device_states.size());
  absl::flat_hash_map<GlobalDeviceId, int> device_to_node;
  for (const LocalTopologyProto& node : global_topology.nodes()) {
    for (const DeviceProto& device_proto : node.devices()) {
      GlobalDeviceId global_device_id(device_proto.global_device_id());
      device_to_node[global_device_id] = node.node_id();
      std::unique_ptr<LocalDeviceState> local_device;
      if (node.node_id() == node_id) {
        TF_RET_CHECK(device_proto.local_device_ordinal() >= 0 &&
                     device_proto.local_device_ordinal() <
                         local_device_states.size());
        TF_RET_CHECK(local_device_states[device_proto.local_device_ordinal()] !=
                     nullptr);
        local_device =
            std::move(local_device_states[device_proto.local_device_ordinal()]);
        gpu_device_ids[device_proto.local_device_ordinal()] = global_device_id;
      }
      auto device = absl::make_unique<GpuDevice>(
          device_proto.global_device_id(), std::move(local_device),
          device_proto.name(), node.node_id());
      devices->push_back(std::move(device));
    }
  }
  for (const auto& device : local_device_states) {
    TF_RET_CHECK(device == nullptr);
  }
  gpu_executable_run_options->set_gpu_global_device_ids(
      std::move(gpu_device_ids));
  auto nccl_id_store = std::make_shared<NcclIdStore>(
      node_id, distributed_client, device_to_node);
  gpu_executable_run_options->set_nccl_unique_id_callback(
      [nccl_id_store](const gpu::NcclCliqueKey& key) {
        return nccl_id_store->GetNcclUniqueId(key);
      });
  return Status::OK();
}

}  // namespace

GpuDevice::GpuDevice(int id,
                     std::unique_ptr<LocalDeviceState> local_device_state,
                     std::string device_kind, int node_id)
    : PjRtStreamExecutorDevice(id, std::move(local_device_state),
                               std::move(device_kind), node_id) {}

StatusOr<std::unique_ptr<PjRtClient>> GetGpuClient(
    bool asynchronous, const GpuAllocatorConfig& allocator_config,
    std::shared_ptr<DistributedRuntimeClient> distributed_client, int node_id) {
  TF_ASSIGN_OR_RETURN(LocalClient * xla_client, GetGpuXlaClient());
  TF_ASSIGN_OR_RETURN(
      std::vector<std::unique_ptr<LocalDeviceState>> local_device_states,
      BuildLocalDeviceStates(xla_client, asynchronous));
  TF_ASSIGN_OR_RETURN(
      auto allocator,
      GetGpuDeviceAllocator(xla_client->platform(), allocator_config,
                            local_device_states));
  auto host_memory_allocator =
      GetGpuHostAllocator(local_device_states.front()->executor());

  std::vector<std::unique_ptr<PjRtStreamExecutorDevice>> devices;
  auto gpu_run_options = absl::make_unique<gpu::GpuExecutableRunOptions>();
  if (distributed_client) {
    TF_RETURN_IF_ERROR(BuildDistributedDevices(
        std::move(local_device_states), std::move(distributed_client), node_id,
        &devices, gpu_run_options.get()));
  } else {
    devices = BuildLocalDevices(std::move(local_device_states));
  }

  return std::unique_ptr<PjRtClient>(std::make_unique<GpuClient>(
      kGpuName, xla_client, std::move(devices),
      /*node_id=*/node_id, std::move(allocator),
      std::move(host_memory_allocator),
      /*should_stage_host_to_device_transfers=*/true,
      /*gpu_run_options=*/std::move(gpu_run_options)));
}

}  // namespace xla


text:
// Code generated by client-gen. DO NOT EDIT.

package v1alpha1

import (
	"time"

	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	types "k8s.io/apimachinery/pkg/types"
	watch "k8s.io/apimachinery/pkg/watch"
	rest "k8s.io/client-go/rest"
	v1alpha1 "k8s.io/sample-controller/pkg/apis/samplecontroller/v1alpha1"
	scheme "k8s.io/sample-controller/pkg/client/clientset/versioned/scheme"
)

// FoosGetter has a method to return a FooInterface.
// A group's client should implement this interface.
type FoosGetter interface {
	Foos(namespace string) FooInterface
}

// FooInterface has methods to work with Foo resources.
type FooInterface interface {
	Create(*v1alpha1.Foo) (*v1alpha1.Foo, error)
	Update(*v1alpha1.Foo) (*v1alpha1.Foo, error)
	UpdateStatus(*v1alpha1.Foo) (*v1alpha1.Foo, error)
	Delete(name string, options *v1.DeleteOptions) error
	DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error
	Get(name string, options v1.GetOptions) (*v1alpha1.Foo, error)
	List(opts v1.ListOptions) (*v1alpha1.FooList, error)
	Watch(opts v1.ListOptions) (watch.Interface, error)
	Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.Foo, err error)
	FooExpansion
}

// foos implements FooInterface
type foos struct {
	client rest.Interface
	ns     string
}

// newFoos returns a Foos
func newFoos(c *SamplecontrollerV1alpha1Client, namespace string) *foos {
	return &foos{
		client: c.RESTClient(),
		ns:     namespace,
	}
}

// Get takes name of the foo, and returns the corresponding foo object, and an error if there is any.
func (c *foos) Get(name string, options v1.GetOptions) (result *v1alpha1.Foo, err error) {
	result = &v1alpha1.Foo{}
	err = c.client.Get().
		Namespace(c.ns).
		Resource("foos").
		Name(name).
		VersionedParams(&options, scheme.ParameterCodec).
		Do().
		Into(result)
	return
}

// List takes label and field selectors, and returns the list of Foos that match those selectors.
func (c *foos) List(opts v1.ListOptions) (result *v1alpha1.FooList, err error) {
	var timeout time.Duration
	if opts.TimeoutSeconds != nil {
		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
	}
	result = &v1alpha1.FooList{}
	err = c.client.Get().
		Namespace(c.ns).
		Resource("foos").
		VersionedParams(&opts, scheme.ParameterCodec).
		Timeout(timeout).
		Do().
		Into(result)
	return
}

// Watch returns a watch.Interface that watches the requested foos.
func (c *foos) Watch(opts v1.ListOptions) (watch.Interface, error) {
	var timeout time.Duration
	if opts.TimeoutSeconds != nil {
		timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
	}
	opts.Watch = true
	return c.client.Get().
		Namespace(c.ns).
		Resource("foos").
		VersionedParams(&opts, scheme.ParameterCodec).
		Timeout(timeout).
		Watch()
}

// Create takes the representation of a foo and creates it.  Returns the server's representation of the foo, and an error, if there is any.
func (c *foos) Create(foo *v1alpha1.Foo) (result *v1alpha1.Foo, err error) {
	result = &v1alpha1.Foo{}
	err = c.client.Post().
		Namespace(c.ns).
		Resource("foos").
		Body(foo).
		Do().
		Into(result)
	return
}

// Update takes the representation of a foo and updates it. Returns the server's representation of the foo, and an error, if there is any.
func (c *foos) Update(foo *v1alpha1.Foo) (result *v1alpha1.Foo, err error) {
	result = &v1alpha1.Foo{}
	err = c.client.Put().
		Namespace(c.ns).
		Resource("foos").
		Name(foo.Name).
		Body(foo).
		Do().
		Into(result)
	return
}

// UpdateStatus was generated because the type contains a Status member.
// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().

func (c *foos) UpdateStatus(foo *v1alpha1.Foo) (result *v1alpha1.Foo, err error) {
	result = &v1alpha1.Foo{}
	err = c.client.Put().
		Namespace(c.ns).
		Resource("foos").
		Name(foo.Name).
		SubResource("status").
		Body(foo).
		Do().
		Into(result)
	return
}

// Delete takes name of the foo and deletes it. Returns an error if one occurs.
func (c *foos) Delete(name string, options *v1.DeleteOptions) error {
	return c.client.Delete().
		Namespace(c.ns).
		Resource("foos").
		Name(name).
		Body(options).
		Do().
		Error()
}

// DeleteCollection deletes a collection of objects.
func (c *foos) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error {
	var timeout time.Duration
	if listOptions.TimeoutSeconds != nil {
		timeout = time.Duration(*listOptions.TimeoutSeconds) * time.Second
	}
	return c.client.Delete().
		Namespace(c.ns).
		Resource("foos").
		VersionedParams(&listOptions, scheme.ParameterCodec).
		Timeout(timeout).
		Body(options).
		Do().
		Error()
}

// Patch applies the patch and returns the patched foo.
func (c *foos) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.Foo, err error) {
	result = &v1alpha1.Foo{}
	err = c.client.Patch(pt).
		Namespace(c.ns).
		Resource("foos").
		SubResource(subresources...).
		Name(name).
		Body(data).
		Do().
		Into(result)
	return
}


text:
/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *	  http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.asual.summer.core.resource;

import org.springframework.core.PriorityOrdered;

/**
 *
 * @author Rostislav Hristov
 *
 */
public abstract class AbstractResource implements PriorityOrdered {

	private int order = 0;

	public int getOrder() {
		return order;
	}

	public void setOrder(int order) {
		this.order = order;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + order;
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		AbstractResource other = (AbstractResource) obj;
		if (order != other.order)
			return false;
		return true;
	}

}


text:
Just some simple code to parse/visualize a few time series datasets. This is mostly just for me, but figured I'd share it for others who may be using the same data.

Currently works with:
- MSRC-12
- PAMAP
- PAMAP2
- The UCR Time Series datasets

To actually run this code, you should just need to change a few directory names in paths.py.


text:
var log4js = require("log4js"),
    log = log4js.getLogger("megapis-worker"),
    FeedParser = require("feedparser"),
    async = require("async"),
    request = require("request"),
    moment = require("moment"),
    feedUtils = require("feedparser/utils"),
    util = require("util");

var MegapisWorker = require("megapis-worker").MegapisWorker;

function RssWorker(config) {
    RssWorker.super_.call(this, config);
}

util.inherits(RssWorker, MegapisWorker);

exports.createWorker = function(config) {
    return new RssWorker(config);
};

RssWorker.prototype.getConfigKeys = function() {
    return ["output", "urls"];
};

function fetchRss(config, callback) {
    var data = [];
    var maxItems = 0; // all
    if (config.maxItemsPerFeed) {
        maxItems = parseInt(config.maxItemsPerFeed);
    }
    var minDate = moment();
    if (config.maxDaysOld) {
        minDate.subtract(config.maxDaysOld, "days");
    } else {
        minDate.subtract(100, "years");
    }
    async.forEachSeries(config.urls,
        function(url, forEachCallback) {
            log.debug("get ", url);
            var meta = {};
            var items = [];
            var feedparser = new FeedParser();
            var req = request(url);
            req.on("error", function(error) {
                log.error("error loading "+url+": ", error);
            });
            req.on("response", function (res) {
                var stream = this;
                if (res.statusCode != 200) return this.emit("error", new Error("Bad status code"));
                stream.pipe(feedparser);
            });
            feedparser.on("error", function(error) {
                log.error("error parsing feed: ", error);
                forEachCallback();
            });
            feedparser.on("readable", function() {
                var stream = this;
                meta = this.meta;
                var item = stream.read();
                while (item) {
                    var dt = moment(item.pubdate);
                    var read = config.excludeUrls && config.excludeUrls.indexOf(item.link) >= 0;
                    log.debug("url="+item.link+" read="+read);
                    if (!read && dt.isAfter(minDate)) {
                        items.push({
                            title: item.title,
                            summary: feedUtils.stripHtml(item.summary),
                            url: item.link,
                            guid: item.guid,
                            image: item.image,
                            pubdate: item.pubdate,
                        });
                    }
                    item = stream.read();
                }
            });
            feedparser.on("end", function() {
                log.debug("done with "+url+"; "+items.length+" items");
                if (items.length) {
                    if (maxItems) {
                        items = items.slice(0, maxItems);
                    }
                    data.push({
                        title: meta.title,
                        url: meta.link,
                        items: items
                    });
                }
                forEachCallback();
            });
        }, function(err) {
            if (err) {
                log.error("error ", err);
            } else {
                log.debug("done "+Object.keys(data).length+" sources");
            }
            callback(err, data);
        });
}

RssWorker.prototype.run = function(callback) {
    var self = this;
    this.config.excludeUrls = [];
    if (this.config.excludeKey) {
        this.get(this.config.excludeKey, function(err, values) {
            self.config.excludeUrls = values;
            fetchRss(self.config, function(err, data) {
                if (!err) {
                    self.save(data, callback);
                }
            });
        });
    } else {
        fetchRss(this.config, function(err, data) {
            if (!err) {
                self.save(data, callback);
            }
        });
    }
};


text:
'use strict';

var fs = require('fs');
var path = require('path');
var _ = require('lodash');
var Q = require('q');
var zlib = require('zlib');

var defaultOptions = {
    tmp: 'tmp',
    gzipLevel: 6
};

module.exports = function(options) {

    _.defaults(options, defaultOptions);

    function getSize(filename) {
        var size = 0;

        if (typeof filename === 'string') {
            try {
                size = fs.statSync(filename).size;
            } catch (e) {}
        } else {
            size = filename;
        }

        return Number(size);
    }

    return function(filename) {
        var deferred = Q.defer();
        var file = path.basename(filename);
        var dest = path.resolve(__dirname, '../../', options.tmp, file + '.gz');

        var srcStream = fs.createReadStream(filename);
        var destStream = fs.createWriteStream(dest);
        var compressor = zlib.createGzip.call(zlib, options);

        compressor.on('error', function() {
            deferred.reject(new Error('GZip compression of ' + filename + ' failed.'));
        });

        destStream.on('close', function() {
            deferred.resolve({
                uncompressed: Number(getSize(filename)),
                compressed: Number(getSize(dest))
            });
        });

        srcStream.pipe(compressor).pipe(destStream);

        return deferred.promise;
    };

};


text:
package org.trustedanalytics.les;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import org.trustedanalytics.les.gathering.EventCollector;
import org.trustedanalytics.les.nats.EventProcessor;
import org.trustedanalytics.les.nats.EventRetriever;
import org.trustedanalytics.les.nats.NatsEventInfo;
import org.trustedanalytics.les.storage.EventInfo;
import org.trustedanalytics.les.storage.EventStore;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;

import java.util.UUID;

@RunWith(MockitoJUnitRunner.class)
public class EventsCollectorTests {

    private EventCollector sut;

    @Mock
    private EventRetriever retriever;

    @Mock
    private EventStore store;

    @Test
    public void constructor_collectorRegistersItselfAsEventProcessor() {
        sut = new EventCollector(retriever, store);

        verify(retriever, times(1)).setEventProcessor(any(EventProcessor.class));
    }

    @Test
    public void process_whenEventArrivesItIsStored() {
        sut = new EventCollector(retriever, store);
        ArgumentCaptor<EventProcessor> processor = ArgumentCaptor.forClass(EventProcessor.class);
        verify(retriever, times(1)).setEventProcessor(processor.capture());

        NatsEventInfo natsEvent = new NatsEventInfo();
        natsEvent.setServiceId("123-abc");
        natsEvent.setOrganizationId(UUID.randomUUID().toString());

        processor.getValue().process("service-creation", natsEvent);

        ArgumentCaptor<EventInfo> event = ArgumentCaptor.forClass(EventInfo.class);
        verify(store, times(1)).save(event.capture());
        assertNotNull(event.getValue());
        assertEquals("123-abc", event.getValue().getSourceId());
    }
}


text:
mementodb
=========

Time series database