Skip to content

Commit

Permalink
Added new storage impl
Browse files Browse the repository at this point in the history
  • Loading branch information
@niemyjski
niemyjski committed Jan 4, 2024
1 parent f27b02d commit 793d51e
Showing 1 changed file with 38 additions and 30 deletions.
68 changes: 38 additions & 30 deletions src/Foundatio.Redis/Storage/RedisFileStorage.cs
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
using System;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
Expand All @@ -22,10 +22,10 @@ public class RedisFileStorage : IFileStorage {
public RedisFileStorage(RedisFileStorageOptions options) {
if (options.ConnectionMultiplexer == null)
throw new ArgumentException("ConnectionMultiplexer is required.");

_serializer = options.Serializer ?? DefaultSerializer.Instance;
_logger = options.LoggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;

options.ConnectionMultiplexer.ConnectionRestored += ConnectionMultiplexerOnConnectionRestored;
_fileSpecContainer = $"{options.ContainerName}-filespecs";
_options = options;
Expand All @@ -41,47 +41,55 @@ public void Dispose() {
_options.ConnectionMultiplexer.ConnectionRestored -= ConnectionMultiplexerOnConnectionRestored;
}

public async Task<Stream> GetFileStreamAsync(string path, CancellationToken cancellationToken = default) {
[Obsolete($"Use {nameof(GetFileStreamAsync)} with {nameof(FileAccess)} instead to define read or write behaviour of stream")]
public Task<Stream> GetFileStreamAsync(string path, CancellationToken cancellationToken = default)
=> GetFileStreamAsync(path, StreamMode.Read, cancellationToken);

public async Task<Stream> GetFileStreamAsync(string path, StreamMode streamMode, CancellationToken cancellationToken = default)

Check failure on line 48 in src/Foundatio.Redis/Storage/RedisFileStorage.cs

View workflow job for this annotation

GitHub Actions / build / build 

The type or namespace name 'StreamMode' could not be found (are you missing a using directive or an assembly reference?)

Check failure on line 48 in src/Foundatio.Redis/Storage/RedisFileStorage.cs

View workflow job for this annotation

GitHub Actions / build / build 

The type or namespace name 'StreamMode' could not be found (are you missing a using directive or an assembly reference?)
{
if (String.IsNullOrEmpty(path))
throw new ArgumentNullException(nameof(path));


if (streamMode is StreamMode.Write)
throw new NotSupportedException($"Stream mode {streamMode} is not supported.");

string normalizedPath = NormalizePath(path);
_logger.LogTrace("Getting file stream for {Path}", normalizedPath);

var fileContent = await Run.WithRetriesAsync(() => Database.HashGetAsync(_options.ContainerName, normalizedPath),
cancellationToken: cancellationToken, logger: _logger).AnyContext();

if (fileContent.IsNull) {
_logger.LogError("Unable to get file stream for {Path}: File Not Found", normalizedPath);
return null;
}

return new MemoryStream(fileContent);
}

public async Task<FileSpec> GetFileInfoAsync(string path) {
if (String.IsNullOrEmpty(path))
throw new ArgumentNullException(nameof(path));

string normalizedPath = NormalizePath(path);
_logger.LogTrace("Getting file info for {Path}", normalizedPath);

var fileSpec = await Run.WithRetriesAsync(() => Database.HashGetAsync(_fileSpecContainer, normalizedPath), logger: _logger).AnyContext();
if (!fileSpec.HasValue) {
_logger.LogError("Unable to get file info for {Path}: File Not Found", normalizedPath);
return null;
}

return _serializer.Deserialize<FileSpec>((byte[])fileSpec);
}

public Task<bool> ExistsAsync(string path) {
if (String.IsNullOrEmpty(path))
throw new ArgumentNullException(nameof(path));

string normalizedPath = NormalizePath(path);
_logger.LogTrace("Checking if {Path} exists", normalizedPath);

return Run.WithRetriesAsync(() => Database.HashExistsAsync(_fileSpecContainer, normalizedPath), logger: _logger);
}

Expand All @@ -90,7 +98,7 @@ public async Task<bool> SaveFileAsync(string path, Stream stream, CancellationTo
throw new ArgumentNullException(nameof(path));
if (stream == null)
throw new ArgumentNullException(nameof(stream));

string normalizedPath = NormalizePath(path);
_logger.LogTrace("Saving {Path}", normalizedPath);

Expand All @@ -103,7 +111,7 @@ public async Task<bool> SaveFileAsync(string path, Stream stream, CancellationTo
long fileSize = memory.Length;
memory.Seek(0, SeekOrigin.Begin);
memory.SetLength(0);

_serializer.Serialize(new FileSpec {
Path = normalizedPath,
Created = DateTime.UtcNow,
Expand All @@ -126,7 +134,7 @@ public async Task<bool> RenameFileAsync(string path, string newPath, Cancellatio
throw new ArgumentNullException(nameof(path));
if (String.IsNullOrEmpty(newPath))
throw new ArgumentNullException(nameof(newPath));

string normalizedPath = NormalizePath(path);
string normalizedNewPath = NormalizePath(newPath);
_logger.LogInformation("Renaming {Path} to {NewPath}", normalizedPath, normalizedNewPath);
Expand All @@ -146,7 +154,7 @@ public async Task<bool> CopyFileAsync(string path, string targetPath, Cancellati
throw new ArgumentNullException(nameof(path));
if (String.IsNullOrEmpty(targetPath))
throw new ArgumentNullException(nameof(targetPath));

string normalizedPath = NormalizePath(path);
string normalizedTargetPath = NormalizePath(targetPath);
_logger.LogInformation("Copying {Path} to {TargetPath}", normalizedPath, normalizedTargetPath);
Expand All @@ -166,7 +174,7 @@ public async Task<bool> CopyFileAsync(string path, string targetPath, Cancellati
public async Task<bool> DeleteFileAsync(string path, CancellationToken cancellationToken = default) {
if (String.IsNullOrEmpty(path))
throw new ArgumentNullException(nameof(path));

string normalizedPath = NormalizePath(path);
_logger.LogTrace("Deleting {Path}", normalizedPath);

Expand All @@ -187,14 +195,14 @@ public async Task<int> DeleteFilesAsync(string searchPattern = null, Cancellatio
count++;
}
_logger.LogTrace("Finished deleting {FileCount} files matching {SearchPattern}", count, searchPattern);

return count;
}

private Task<List<FileSpec>> GetFileListAsync(string searchPattern = null, int? limit = null, int? skip = null, CancellationToken cancellationToken = default) {
if (limit is <= 0)
return Task.FromResult(new List<FileSpec>());

searchPattern = NormalizePath(searchPattern);
string prefix = searchPattern;
Regex patternRegex = null;
Expand All @@ -204,15 +212,15 @@ private Task<List<FileSpec>> GetFileListAsync(string searchPattern = null, int?
int slashPos = searchPattern.LastIndexOf('/');
prefix = slashPos >= 0 ? searchPattern.Substring(0, slashPos) : String.Empty;
}

prefix ??= String.Empty;
int pageSize = limit ?? Int32.MaxValue;

_logger.LogTrace(
s => s.Property("SearchPattern", searchPattern).Property("Limit", limit).Property("Skip", skip),
s => s.Property("SearchPattern", searchPattern).Property("Limit", limit).Property("Skip", skip),
"Getting file list matching {Prefix} and {Pattern}...", prefix, patternRegex
);

return Task.FromResult(Database.HashScan(_fileSpecContainer, $"{prefix}*")
.Select(entry => _serializer.Deserialize<FileSpec>((byte[])entry.Value))
.Where(fileSpec => patternRegex == null || patternRegex.IsMatch(fileSpec.Path))
Expand All @@ -236,12 +244,12 @@ private NextPageResult GetFiles(SearchCriteria criteria, int page, int pageSize)
int skip = (page - 1) * pagingLimit;
if (pagingLimit < Int32.MaxValue)
pagingLimit++;

_logger.LogTrace(
s => s.Property("Limit", pagingLimit).Property("Skip", skip),
s => s.Property("Limit", pagingLimit).Property("Skip", skip),
"Getting files matching {Prefix} and {Pattern}...", criteria.Prefix, criteria.Pattern
);

var list = Database.HashScan(_fileSpecContainer, $"{criteria.Prefix}*")
.Select(entry => _serializer.Deserialize<FileSpec>((byte[])entry.Value))
.Where(fileSpec => criteria.Pattern == null || criteria.Pattern.IsMatch(fileSpec.Path))
Expand All @@ -266,7 +274,7 @@ private NextPageResult GetFiles(SearchCriteria criteria, int page, int pageSize)
private string NormalizePath(string path) {
return path?.Replace('\\', '/');
}

private class SearchCriteria {
public string Prefix { get; set; }
public Regex Pattern { get; set; }
Expand All @@ -275,14 +283,14 @@ private class SearchCriteria {
private SearchCriteria GetRequestCriteria(string searchPattern) {
if (String.IsNullOrEmpty(searchPattern))
return new SearchCriteria { Prefix = String.Empty };

string normalizedSearchPattern = NormalizePath(searchPattern);
int wildcardPos = normalizedSearchPattern.IndexOf('*');
bool hasWildcard = wildcardPos >= 0;

string prefix = normalizedSearchPattern;
Regex patternRegex = null;

if (hasWildcard) {
patternRegex = new Regex($"^{Regex.Escape(normalizedSearchPattern).Replace("\\*", ".*?")}$");
int slashPos = normalizedSearchPattern.LastIndexOf('/');
Expand Down

0 comments on commit 793d51e

Please sign in to comment.