package org.apache.cassandra.hints;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Stream;
import javax.annotation.Nullable;
import com.google.common.collect.ImmutableMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.NativeLibrary;
import org.apache.cassandra.utils.SyncUtil;
import static java.util.stream.Collectors.groupingBy;
final class HintsCatalog
{
private static final Logger logger = LoggerFactory.getLogger(HintsCatalog.class);
private final File hintsDirectory;
private final Map<UUID, HintsStore> stores;
private final ImmutableMap<String, Object> writerParams;
private HintsCatalog(File hintsDirectory, ImmutableMap<String, Object> writerParams, Map<UUID, List<HintsDescriptor>> descriptors)
{
this.hintsDirectory = hintsDirectory;
this.writerParams = writerParams;
this.stores = new ConcurrentHashMap<>();
for (Map.Entry<UUID, List<HintsDescriptor>> entry : descriptors.entrySet())
stores.put(entry.getKey(), HintsStore.create(entry.getKey(), hintsDirectory, writerParams, entry.getValue()));
}
static HintsCatalog load(File hintsDirectory, ImmutableMap<String, Object> writerParams)
{
try(Stream<Path> list = Files.list(hintsDirectory.toPath()))
{
Map<UUID, List<HintsDescriptor>> stores =
list
.filter(HintsDescriptor::isHintFileName)
.map(HintsDescriptor::readFromFileQuietly)
.filter(Optional::isPresent)
.map(Optional::get)
.collect(groupingBy(h -> h.hostId));
return new HintsCatalog(hintsDirectory, writerParams, stores);
}
catch (IOException e)
{
throw new FSReadError(e, hintsDirectory);
}
}
Stream<HintsStore> stores()
{
return stores.values().stream();
}
void maybeLoadStores(Iterable<UUID> hostIds)
{
for (UUID hostId : hostIds)
get(hostId);
}
HintsStore get(UUID hostId)
{
HintsStore store = stores.get(hostId);
return store == null
? stores.computeIfAbsent(hostId, (id) -> HintsStore.create(id, hintsDirectory, writerParams, Collections.emptyList()))
: store;
}
@Nullable
HintsStore getNullable(UUID hostId)
{
return stores.get(hostId);
}
void deleteAllHints()
{
stores.keySet().forEach(this::deleteAllHints);
}
void deleteAllHints(UUID hostId)
{
HintsStore store = stores.get(hostId);
if (store != null)
store.deleteAllHints();
}
boolean hasFiles()
{
return stores().anyMatch(HintsStore::hasFiles);
}
void exciseStore(UUID hostId)
{
deleteAllHints(hostId);
stores.remove(hostId);
}
void fsyncDirectory()
{
int fd = NativeLibrary.tryOpenDirectory(hintsDirectory.getAbsolutePath());
if (fd != -1)
{
try
{
SyncUtil.trySync(fd);
NativeLibrary.tryCloseFD(fd);
}
catch (FSError e)
{
logger.error("Unable to sync directory {}", hintsDirectory.getAbsolutePath(), e);
FileUtils.handleFSErrorAndPropagate(e);
}
}
else
{
logger.error("Unable to open directory {}", hintsDirectory.getAbsolutePath());
FileUtils.handleFSErrorAndPropagate(new FSWriteError(new IOException(String.format("Unable to open hint directory %s", hintsDirectory.getAbsolutePath())), hintsDirectory.getAbsolutePath()));
}
}
ImmutableMap<String, Object> getWriterParams()
{
return writerParams;
}
}