import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Constructs new shared file system repository
 *
 * @param name                 repository name
 * @param repositorySettings   repository settings
 * @param indexShardRepository index shard repository
 */
@Inject
public FsRepository(RepositoryName name, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, Environment environment) throws IOException {
    super(name.getName(), repositorySettings, indexShardRepository);
    Path locationFile;
    String location = repositorySettings.settings().get("location", settings.get("repositories.fs.location"));
    if (location == null) {
        logger.warn("the repository location is missing, it should point to a shared file system location that is available on all master and data nodes");
        throw new RepositoryException(name.name(), "missing location");
    }
    locationFile = environment.resolveRepoFile(location);
    if (locationFile == null) {
        if (environment.repoFiles().length > 0) {
            logger.warn("The specified location [{}] doesn't start with any repository paths specified by the path.repo setting: [{}] ", location, environment.repoFiles());
            throw new RepositoryException(name.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo");
        } else {
            logger.warn("The specified location [{}] should start with a repository path specified by the path.repo setting, but the path.repo setting was not set on this node", location);
            throw new RepositoryException(name.name(), "location [" + location + "] doesn't match any of the locations specified by path.repo because this setting is empty");
        }
    }

    blobStore = new FsBlobStore(settings, locationFile);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", settings.getAsBytesSize("repositories.fs.chunk_size", null));
    this.compress = repositorySettings.settings().getAsBoolean("compress", settings.getAsBoolean("repositories.fs.compress", false));
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Constructs new BlobStoreRepository
 * @param name                 repository name
 * @param repositorySettings   repository settings
 * @param indexShardRepository an instance of IndexShardRepository
 * @param gridFsService and instance of GridFsService
 */
@Inject
protected GridFsRepository(RepositoryName name, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, GridFsService gridFsService) {
    super(name.getName(), repositorySettings, indexShardRepository);

    String database = repositorySettings.settings().get("database", componentSettings.get("database"));
    if (database == null) {
        throw new RepositoryException(name.name(), "No database defined for GridFS repository");
    }

    String bucket = repositorySettings.settings().get("bucket", "fs");
    String host = repositorySettings.settings().get("gridfs_host", "localhost");
    int port = repositorySettings.settings().getAsInt("gridfs_port", 27017);
    String username = repositorySettings.settings().get("gridfs_username");
    String password = repositorySettings.settings().get("gridfs_password");

    int concurrentStreams = repositorySettings.settings().getAsInt("concurrent_streams", componentSettings.getAsInt("concurrent_streams", 5));
    ExecutorService concurrentStreamPool = EsExecutors.newScaling(1, concurrentStreams, 5, TimeUnit.SECONDS, EsExecutors.daemonThreadFactory(settings, "[gridfs_stream]"));
    blobStore = new GridFsBlobStore(settings, gridFsService.mongoDB(host, port, database, username, password), bucket, concurrentStreamPool);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", componentSettings.getAsBytesSize("chunk_size", null));
    this.compress = repositorySettings.settings().getAsBoolean("compress", componentSettings.getAsBoolean("compress", true));
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
@Inject
public SshRepository(final RepositoryName name,
    final RepositorySettings repositorySettings,
    final IndexShardRepository indexShardRepository,
    final ThreadPool threadPool) throws IOException {
    super(name.getName(), repositorySettings, indexShardRepository);

    try {
        blobStore = new SshBlobStore(settings, new JSchClient(
            settings, repositorySettings, threadPool));
    } catch (final JSchException e) {
        throw new RepositoryException(name.name(),
            "Failed to initialize SSH configuration.", e);
    }

    chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size",
        settings.getAsBytesSize("chunk_size", null));
    compress = repositorySettings.settings().getAsBoolean("compress",
        settings.getAsBoolean("compress", false));
    basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Constructs new BlobStoreRepository
 *
 * @param repositoryName       repository name
 * @param repositorySettings   repository settings
 * @param indexShardRepository an instance of IndexShardRepository
 */
protected BlobStoreRepository(String repositoryName, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository) {
    super(repositorySettings.globalSettings());
    this.repositoryName = repositoryName;
    this.indexShardRepository = (BlobStoreIndexShardRepository) indexShardRepository;
    snapshotRateLimiter = getRateLimiter(repositorySettings, "max_snapshot_bytes_per_sec", new ByteSizeValue(40, ByteSizeUnit.MB));
    restoreRateLimiter = getRateLimiter(repositorySettings, "max_restore_bytes_per_sec", new ByteSizeValue(40, ByteSizeUnit.MB));
    readOnly = repositorySettings.settings().getAsBoolean("readonly", false);
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Restores shard from {@link RestoreSource} associated with this shard in routing table
 *
 * @param recoveryState recovery state
 */
private void restore(final IndexShard indexShard, final RecoveryState recoveryState) {
    RestoreSource restoreSource = indexShard.routingEntry().restoreSource();
    if (restoreSource == null) {
        throw new IndexShardRestoreFailedException(shardId, "empty restore source");
    }
    if (logger.isTraceEnabled()) {
        logger.trace("[{}] restoring shard  [{}]", restoreSource.snapshotId(), shardId);
    }
    try {
        recoveryState.getTranslog().totalOperations(0);
        recoveryState.getTranslog().totalOperationsOnStart(0);
        indexShard.prepareForIndexRecovery();
        IndexShardRepository indexShardRepository = repositoriesService.indexShardRepository(restoreSource.snapshotId().getRepository());
        ShardId snapshotShardId = shardId;
        if (!shardId.getIndex().equals(restoreSource.index())) {
            snapshotShardId = new ShardId(restoreSource.index(), shardId.id());
        }
        indexShardRepository.restore(restoreSource.snapshotId(), restoreSource.version(), shardId, snapshotShardId, recoveryState);
        indexShard.skipTranslogRecovery();
        indexShard.finalizeRecovery();
        indexShard.postRecovery("restore done");
        restoreService.indexShardRestoreCompleted(restoreSource.snapshotId(), shardId);
    } catch (Throwable t) {
        if (Lucene.isCorruptionException(t)) {
            restoreService.failRestore(restoreSource.snapshotId(), shardId());
        }
        throw new IndexShardRestoreFailedException(shardId, "restore failed", t);
    }
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
@Inject
protected CloudFilesRepository(String repositoryName, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, CloudFilesService cloudFilesService) {
    super(repositoryName, repositorySettings, indexShardRepository);

    final String container = repositorySettings.settings().get("container", componentSettings.get("container"));
    if(container == null || container.equals("")){
        throw new RepositoryException(repositoryName, "No container defined for cloud files gateway.");
    }

    final String dataCenter = repositorySettings.settings().get("region", componentSettings.get("region", "ORD"));
    final int concurrentStreams = repositorySettings.settings().getAsInt("concurrent_streams", componentSettings.getAsInt("concurrent_streams", 5));
    ExecutorService concurrentStreamPool = EsExecutors.newScaling(1, concurrentStreams, 5, TimeUnit.SECONDS, EsExecutors.daemonThreadFactory(settings, "[cloudfiles_stream]"));
    final Location location = new LocationBuilder()
            .scope(LocationScope.REGION)
            .id(dataCenter)
            .description(String.format("Rackspace's %s datacenter.", dataCenter))
            .build();

    blobStore = new CloudFilesBlobStore(settings, cloudFilesService.context(), container, location, concurrentStreamPool);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size", componentSettings.getAsBytesSize("chunk_size", new ByteSizeValue(100, ByteSizeUnit.MB)));
    this.compress = repositorySettings.settings().getAsBoolean("compress", componentSettings.getAsBoolean("compress", false));
    logger.debug("using container [{}], data center [{}], chunk_size [{}], concurrent_streams [{}]", container, dataCenter, chunkSize, concurrentStreams);
    String basePath = repositorySettings.settings().get("base_path", null);
    if (Strings.hasLength(basePath)) {
        BlobPath path = new BlobPath();
        for(String elem : Strings.splitStringToArray(basePath, '/')) {
            path = path.add(elem);
        }
        this.basePath = path;
    } else {
        this.basePath = BlobPath.cleanPath();
    }
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Constructs new BlobStoreRepository
 * 
 * @param name
 *            repository name
 * @param repositorySettings
 *            repository settings
 * @param indexShardRepository
 *            an instance of IndexShardRepository
 * @param swiftService
 *            an instance of SwiftService
 */
@Inject
public SwiftRepository(RepositoryName name, RepositorySettings repositorySettings, IndexShardRepository indexShardRepository, SwiftService swiftService) {
    super(name.getName(), repositorySettings, indexShardRepository);

    String url = repositorySettings.settings().get("swift_url");
    if (url == null) {
        throw new RepositoryException(name.name(), "No url defined for swift repository");
    }

    String container = repositorySettings.settings().get("swift_container");
    if (container == null) {
        throw new RepositoryException(name.name(), "No container defined for swift repository");
    }

    String username = repositorySettings.settings().get("swift_username", "");
    String password = repositorySettings.settings().get("swift_password", "");
    String tenantName = repositorySettings.settings().get("swift_tenantname", "");
    String authMethod = repositorySettings.settings().get("swift_authmethod", "");
    String preferredRegion = repositorySettings.settings().get("swift_preferred_region", null);
    Account account = SwiftAccountFactory.createAccount(swiftService, url, username, password, tenantName, authMethod, preferredRegion);

    blobStore = new SwiftBlobStore(settings, account, container);
    this.chunkSize = repositorySettings.settings().getAsBytesSize("chunk_size",
            settings.getAsBytesSize("chunk_size", new ByteSizeValue(5, ByteSizeUnit.GB)));
    this.compress = repositorySettings.settings().getAsBoolean("compress",
            settings.getAsBoolean("compress", false));
    this.basePath = BlobPath.cleanPath();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Do the binding.
 */
@Override
protected void configure() {
    bind(Repository.class).to(SwiftRepository.class).asEagerSingleton();
    bind(IndexShardRepository.class).to(BlobStoreIndexShardRepository.class).asEagerSingleton();
    bind(SwiftService.class).asEagerSingleton();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
protected void configure() {
    bind(Repository.class).to(SshRepository.class).asEagerSingleton();
    bind(IndexShardRepository.class)
            .to(BlobStoreIndexShardRepository.class).asEagerSingleton();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/** Registers a custom repository type to the given {@link Repository} and {@link IndexShardRepository}. */
public void registerRepository(String type, Class<? extends Repository> repositoryType, Class<? extends IndexShardRepository> shardRepositoryType) {
    repositoryTypes.registerRepository(type, repositoryType, shardRepositoryType);
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
public RepositoryHolder(String type, Settings settings, Injector injector, Repository repository, IndexShardRepository indexShardRepository) {
    this.type = type;
    this.settings = settings;
    this.repository = repository;
    this.indexShardRepository = indexShardRepository;
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/** Adds a new repository type to the registry, bound to the given implementation classes. */
public void registerRepository(String name, Class<? extends Repository> repositoryType, Class<? extends IndexShardRepository> shardRepositoryType) {
    repositoryTypes.registerExtension(name, repositoryType);
    shardRepositoryTypes.registerExtension(name, shardRepositoryType);
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
private void doVerify(String repository, String verificationToken) {
    IndexShardRepository blobStoreIndexShardRepository = repositoriesService.indexShardRepository(repository);
    blobStoreIndexShardRepository.verify(verificationToken);
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
@Override
protected void configure() {
    bind(Repository.class).to(CloudFilesRepository.class).asEagerSingleton();
    bind(IndexShardRepository.class).to(BlobStoreIndexShardRepository.class).asEagerSingleton();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
@Override
protected void configure() {
    bind(Repository.class).to(GridFsRepository.class).asEagerSingleton();
    bind(IndexShardRepository.class).to(BlobStoreIndexShardRepository.class).asEagerSingleton();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
@Override
protected void configure() {
    bind(WebdavService.class).to(getWebdavServiceClass()).asEagerSingleton();
    bind(Repository.class).to(WebdavRepository.class).asEagerSingleton();
    bind(IndexShardRepository.class).to(BlobStoreIndexShardRepository.class).asEagerSingleton();
}
 

import org.elasticsearch.index.snapshots.IndexShardRepository; //导入依赖的package包/类
/**
 * Returns registered index shard repository
 * <p>
 * This method is called only on data nodes
 *
 * @param repository repository name
 * @return registered repository
 * @throws RepositoryMissingException if repository with such name isn't registered
 */
public IndexShardRepository indexShardRepository(String repository) {
    RepositoryHolder holder = repositories.get(repository);
    if (holder != null) {
        return holder.indexShardRepository;
    }
    throw new RepositoryMissingException(repository);
}