Skip to content
This repository was archived by the owner on Sep 6, 2023. It is now read-only.

Commit 85e4edd

Browse files
seguinsseguins
authored
Merge pull request #39 from LuccaSA/async
Add command to auto-decompress
2 parents 0eb67a1 + 3a2b8f3 commit 85e4edd

6 files changed

Lines changed: 161 additions & 18 deletions

File tree

‎TCC.Lib/Helpers/GoogleAuthHelper.cs‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
using Google.Apis.Auth.OAuth2;
2+
using Google.Cloud.Storage.V1;
3+
using System;
4+
using System.IO;
5+
using System.Text;
6+
using System.Threading;
7+
using System.Threading.Tasks;
8+
9+
namespace TCC.Lib.Helpers
10+
{
11+
public static class GoogleAuthHelper
12+
{
13+
public static async Task<GoogleCredential> GetGoogleClientAsync(string googleStorageCredential, CancellationToken token)
14+
{
15+
if (File.Exists(googleStorageCredential))
16+
{
17+
return await GoogleCredential.FromFileAsync(googleStorageCredential, token);
18+
}
19+
else
20+
{
21+
var decodedJson = Encoding.UTF8.GetString(Convert.FromBase64String(googleStorageCredential));
22+
return GoogleCredential.FromJson(decodedJson);
23+
}
24+
}
25+
26+
public static async Task<StorageClient> GetGoogleStorageClientAsync(string googleStorageCredential, CancellationToken token)
27+
{
28+
var credential = await GetGoogleClientAsync(googleStorageCredential, token);
29+
return await StorageClient.CreateAsync(credential);
30+
}
31+
32+
}
33+
}

‎TCC.Lib/Storage/RemoteStorageFactory.cs‎

Lines changed: 2 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
using System.Text;
88
using System.Threading;
99
using System.Threading.Tasks;
10+
using TCC.Lib.Helpers;
1011
using TCC.Lib.Options;
1112

1213
namespace TCC.Lib.Storage
@@ -38,7 +39,7 @@ public static async Task<IRemoteStorage> GetRemoteStorageAsync(this CompressOpti
3839
logger.LogCritical("Configuration error for google storage upload");
3940
return new NoneRemoteStorage();
4041
}
41-
StorageClient storage = await GetGoogleStorageClient(option, token);
42+
StorageClient storage = await GoogleAuthHelper.GetGoogleStorageClientAsync(option.GoogleStorageCredential, token);
4243
return new GoogleRemoteStorage(storage, option.GoogleStorageBucketName);
4344
}
4445
case UploadMode.None:
@@ -48,20 +49,5 @@ public static async Task<IRemoteStorage> GetRemoteStorageAsync(this CompressOpti
4849
throw new ArgumentOutOfRangeException();
4950
}
5051
}
51-
52-
private static async Task<StorageClient> GetGoogleStorageClient(CompressOption option, CancellationToken token)
53-
{
54-
GoogleCredential credential;
55-
if (File.Exists(option.GoogleStorageCredential))
56-
{
57-
credential = await GoogleCredential.FromFileAsync(option.GoogleStorageCredential, token);
58-
}
59-
else
60-
{
61-
var decodedJson = Encoding.UTF8.GetString(Convert.FromBase64String(option.GoogleStorageCredential));
62-
credential = GoogleCredential.FromJson(decodedJson);
63-
}
64-
return await StorageClient.CreateAsync(credential);
65-
}
6652
}
6753
}

‎TCC/Parser/AutoDecompressCommand.cs‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
using System.Collections.Generic;
2+
using System.CommandLine;
3+
using System.Threading.Tasks;
4+
using TCC.Lib.Options;
5+
6+
namespace TCC.Parser
7+
{
8+
public class AutoDecompressOptionBinding : DecompressOption
9+
{
10+
public AutoDecompressOptionBinding()
11+
{
12+
base.Threads = 1;
13+
}
14+
public new string Threads { set => base.Threads = ParseCommandLineHelper.ThreadsParsing(value); }
15+
public string Password { set => ParseCommandLineHelper.ExtractInlinePassword(this, value); }
16+
public string PassFile { set => ParseCommandLineHelper.ExtractPasswordFile(this, value); }
17+
public string Key { set => ParseCommandLineHelper.ExtractAsymetricFile(this, Mode.Decompress, value); }
18+
public string GoogleStorageCredential { get; set; }
19+
public string GoogleProjectId { get; set; }
20+
public string GoogleSubscriptionId { get; set; }
21+
public string TemporaryDirectory { get; set; }
22+
}
23+
24+
25+
public class AutoDecompressCommand : TccCommand<AutoDecompressOptionBinding>
26+
{
27+
public AutoDecompressCommand() : base("auto-decompress", "Continuous decompress from Google Cloud Storage")
28+
{
29+
}
30+
31+
protected override IEnumerable<Argument> CreateArguments()
32+
{
33+
yield break;
34+
}
35+
protected override IEnumerable<Option> CreateOptions()
36+
{
37+
foreach (var option in BaseCmdOptions.CreateBaseOptions())
38+
{
39+
yield return option;
40+
}
41+
yield return new Option<string>(new[] { "--googleStorageCredential" }, "Google Cloud Storage credential json, either full path or base64");
42+
yield return new Option<string>(new[] { "--googleProjectId" }, "Google Cloud Pub/Sub, storage projectId");
43+
yield return new Option<string>(new[] { "--googleSubscriptionId" }, "Google Cloud Pub/Sub subscriptionId");
44+
yield return new Option<string>(new[] { "--temporaryDirectory" }, "Directory where archive are downloaded temporary");
45+
}
46+
47+
protected override Task RunAsync(ITccController controller, AutoDecompressOptionBinding option)
48+
=> controller.AutoDecompressAsync(option);
49+
}
50+
}

‎TCC/Program.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ await BuildCommandLine()
4848
public static CommandLineBuilder BuildCommandLine()
4949
=> new CommandLineBuilder(new RootCommand
5050
{
51+
new AutoDecompressCommand(),
5152
new CompressCommand(),
5253
new DecompressCommand(),
5354
new BenchmarkCommand()

‎TCC/TCC.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
</PropertyGroup>
1010

1111
<ItemGroup>
12+
<PackageReference Include="Google.Cloud.PubSub.V1" Version="2.10.0" />
1213
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="6.0.0" />
1314
<PackageReference Include="Microsoft.Extensions.Logging" Version="6.0.0" />
1415
<PackageReference Include="Serilog.AspNetCore" Version="5.0.0" />

‎TCC/TccController.cs‎

Lines changed: 74 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,22 @@
1-
using Microsoft.Extensions.Logging;
1+
using Google.Apis.Auth.OAuth2;
2+
using Google.Cloud.PubSub.V1;
3+
using Google.Cloud.Storage.V1;
4+
using Grpc.Auth;
5+
using Microsoft.Extensions.Logging;
26
using System;
37
using System.Collections.Generic;
8+
using System.IO;
49
using System.Linq;
10+
using System.Text.Json;
11+
using System.Threading;
512
using System.Threading.Tasks;
613
using TCC.Lib;
714
using TCC.Lib.Benchmark;
815
using TCC.Lib.Database;
916
using TCC.Lib.Helpers;
1017
using TCC.Lib.Notification;
1118
using TCC.Lib.Options;
19+
using TCC.Parser;
1220

1321
namespace TCC;
1422

@@ -17,6 +25,7 @@ public interface ITccController
1725
Task CompressAsync(CompressOption option);
1826
Task DecompressAsync(DecompressOption option);
1927
Task BenchmarkAsync(BenchmarkOption option);
28+
Task AutoDecompressAsync(AutoDecompressOptionBinding option);
2029
}
2130

2231
public class TccController : ITccController
@@ -61,6 +70,70 @@ public async Task BenchmarkAsync(BenchmarkOption option)
6170
await LogResultAsync(operationResult, Mode.Benchmark, null);
6271
}
6372

73+
public async Task AutoDecompressAsync(AutoDecompressOptionBinding option)
74+
{
75+
var gcpCredential = await GoogleAuthHelper.GetGoogleClientAsync(option.GoogleStorageCredential, new CancellationToken());
76+
var subscriber = await GetGoogleClientAsync(gcpCredential, option);
77+
var storage = await StorageClient.CreateAsync(gcpCredential);
78+
79+
await _databaseSetup.EnsureDatabaseExistsAsync(Mode.Decompress);
80+
81+
// Use the client as you'd normally do, to listen for messages in this example.
82+
await subscriber.StartAsync(async (msg, cancellationToken) =>
83+
{
84+
if (!msg.Attributes.Any(kvp => kvp.Key == "eventType" && kvp.Value == "OBJECT_FINALIZE"))
85+
{
86+
_logger.LogDebug("EventType not found : {attributes}", msg.Attributes);
87+
return SubscriberClient.Reply.Ack;
88+
}
89+
var msgData = JsonSerializer.Deserialize<ObjectStorageEvent>(msg.Data.ToStringUtf8(), new JsonSerializerOptions { PropertyNameCaseInsensitive = true });
90+
if (string.IsNullOrEmpty(msgData.Name))
91+
{
92+
_logger.LogDebug("Object name not found");
93+
return SubscriberClient.Reply.Ack;
94+
}
95+
if (string.IsNullOrEmpty(msgData.Bucket))
96+
{
97+
_logger.LogDebug("Bucket name not found");
98+
return SubscriberClient.Reply.Ack;
99+
}
100+
101+
var fileName = Path.GetFileName(msgData.Name);
102+
string tempPath;
103+
if (string.IsNullOrEmpty(option.TemporaryDirectory))
104+
{
105+
tempPath = Path.Join(option.TemporaryDirectory, fileName);
106+
}
107+
else
108+
{
109+
tempPath = Path.GetRandomFileName();
110+
}
111+
using (var outputFile = File.OpenWrite(tempPath))
112+
{
113+
await storage.DownloadObjectAsync(msgData.Bucket, msgData.Name, outputFile);
114+
}
115+
_logger.LogDebug("{fileName} downloaded in {path}", fileName, tempPath);
116+
117+
option.SourceDirOrFile = tempPath;
118+
var operationResult = await _tarCompressCrypt.DecompressAsync(option);
119+
await _databaseSetup.CleanupDatabaseAsync(Mode.Decompress);
120+
await LogResultAsync(operationResult, Mode.Decompress, option);
121+
File.Delete(tempPath);
122+
123+
return SubscriberClient.Reply.Ack;
124+
});
125+
}
126+
private record ObjectStorageEvent(string Bucket, string Name);
127+
private static async Task<SubscriberClient> GetGoogleClientAsync(GoogleCredential credential, AutoDecompressOptionBinding option)
128+
{
129+
var subscriptionName = new SubscriptionName(option.GoogleProjectId, option.GoogleSubscriptionId);
130+
// Create a google cloud pub/sub client that reads messages one by one
131+
return await SubscriberClient.CreateAsync(
132+
subscriptionName,
133+
new SubscriberClient.ClientCreationSettings(clientCount: 1, credentials: credential.ToChannelCredentials()),
134+
new SubscriberClient.Settings { FlowControlSettings = new Google.Api.Gax.FlowControlSettings(1, null) }
135+
);
136+
}
64137

65138
private async Task LogResultAsync(OperationSummary operationResult, Mode mode, ISlackOption slackOption)
66139
{
@@ -142,5 +215,4 @@ private void WriteAuditFile(Mode mode, OperationSummary op)
142215
}
143216
}
144217
}
145-
146218
}

0 commit comments

Comments
 (0)