forked from minio/minio-dotnet
-
Notifications
You must be signed in to change notification settings - Fork 0
/
BucketOperations.cs
340 lines (301 loc) · 15.9 KB
/
BucketOperations.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
/*
* Minio .NET Library for Amazon S3 Compatible Cloud Storage, (C) 2017 Minio, Inc.
*
* 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.
*/
using System;
using System.Collections.Generic;
using System.Net;
using System.Threading.Tasks;
using Minio.DataModel;
using RestSharp;
using System.IO;
using System.Linq;
using System.Xml.Linq;
using System.Xml.Serialization;
using Minio.Exceptions;
using System.Globalization;
using System.Reactive.Linq;
using System.Threading;
using System.Text;
namespace Minio
{
public partial class MinioClient : IBucketOperations
{
/// <summary>
/// List all objects in a bucket
/// </summary>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>Task with an iterator lazily populated with objects</returns>
public async Task<ListAllMyBucketsResult> ListBucketsAsync(CancellationToken cancellationToken = default(CancellationToken))
{
var request= await this.CreateRequest(Method.GET,resourcePath:"/").ConfigureAwait(false);
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
ListAllMyBucketsResult bucketList = new ListAllMyBucketsResult();
if (HttpStatusCode.OK.Equals(response.StatusCode))
{
var contentBytes = System.Text.Encoding.UTF8.GetBytes(response.Content);
using (var stream = new MemoryStream(contentBytes))
bucketList = (ListAllMyBucketsResult)(new XmlSerializer(typeof(ListAllMyBucketsResult)).Deserialize(stream));
return bucketList;
}
return bucketList;
}
/// <summary>
/// Create a private bucket with the given name.
/// </summary>
/// <param name="bucketName">Name of the new bucket</param>
/// <param name="location">Region</param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns> Task </returns>
public async Task MakeBucketAsync(string bucketName, string location = "us-east-1", CancellationToken cancellationToken = default(CancellationToken))
{
if (location.Equals("us-east-1"))
{
if (this.Region != "")
{
location = this.Region;
}
}
// Set Target URL
Uri requestUrl = RequestUtil.MakeTargetURL(this.BaseUrl, this.Secure,location);
SetTargetURL(requestUrl);
var request = new RestRequest("/" + bucketName, Method.PUT);
request.XmlSerializer = new RestSharp.Serializers.DotNetXmlSerializer();
request.RequestFormat = DataFormat.Xml;
// ``us-east-1`` is not a valid location constraint according to amazon, so we skip it.
if (location != "us-east-1")
{
CreateBucketConfiguration config = new CreateBucketConfiguration(location);
request.AddBody(config);
}
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// Returns true if the specified bucketName exists, otherwise returns false.
/// </summary>
/// <param name="bucketName">Bucket to test existence of</param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>Task that returns true if exists and user has access</returns>
public async Task<bool> BucketExistsAsync(string bucketName, CancellationToken cancellationToken = default(CancellationToken))
{
try
{
var request = await this.CreateRequest(Method.HEAD, bucketName).ConfigureAwait(false);
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex)
{
if (ex.GetType() == typeof(BucketNotFoundException))
{
return false;
}
throw;
}
return true;
}
/// <summary>
/// Remove a bucket
/// </summary>
/// <param name="bucketName">Name of bucket to remove</param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>Task</returns>
public async Task RemoveBucketAsync(string bucketName, CancellationToken cancellationToken = default(CancellationToken))
{
var request = await this.CreateRequest(Method.DELETE, bucketName, resourcePath: null).ConfigureAwait(false);
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// List all objects non-recursively in a bucket with a given prefix, optionally emulating a directory
/// </summary>
/// <param name="bucketName">Bucket to list objects from</param>
/// <param name="prefix">Filters all objects beginning with a given prefix</param>
/// <param name="recursive">Set to true to recursively list all objects</param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>An observable of items that client can subscribe to</returns>
public IObservable<Item> ListObjectsAsync(string bucketName, string prefix = "", bool recursive = false, CancellationToken cancellationToken = default(CancellationToken))
{
return Observable.Create<Item>(
async obs =>
{
bool isRunning = true;
string marker = null;
var delimiter = "/";
if (recursive)
{
delimiter = "";
}
while (isRunning)
{
Tuple<ListBucketResult, List<Item>> result = await GetObjectListAsync(bucketName, prefix, delimiter, marker, cancellationToken).ConfigureAwait(false);
Item lastItem = null;
foreach (Item item in result.Item2)
{
lastItem = item;
obs.OnNext(item);
}
if (result.Item1.NextMarker != null)
{
marker = result.Item1.NextMarker;
}
else if (lastItem != null)
{
marker = lastItem.Key;
}
isRunning = result.Item1.IsTruncated;
}
});
}
/// <summary>
/// Gets the list of objects in the bucket filtered by prefix
/// </summary>
/// <param name="bucketName">Bucket to list objects from</param>
/// <param name="prefix">Filters all objects starting with a given prefix</param>
/// <param name="delimiter">Delimit the output upto this character</param>
/// <param name="marker">marks location in the iterator sequence</param>
/// <returns>Task with a tuple populated with objects</returns>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
private async Task<Tuple<ListBucketResult, List<Item>>> GetObjectListAsync(string bucketName, string prefix, string delimiter, string marker, CancellationToken cancellationToken = default(CancellationToken))
{
var queries = new List<string>();
queries.Add("delimiter="+ Uri.EscapeDataString(delimiter));
queries.Add("prefix=" + Uri.EscapeDataString(prefix));
queries.Add("max-keys=1000");
if (marker != null)
{
queries.Add("marker=" + Uri.EscapeDataString(marker));
}
string query = string.Join("&", queries);
var request = await this.CreateRequest(Method.GET,
bucketName,
resourcePath: "?" + query)
.ConfigureAwait(false);
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
var contentBytes = System.Text.Encoding.UTF8.GetBytes(response.Content);
ListBucketResult listBucketResult = null;
using (var stream = new MemoryStream(contentBytes))
{
listBucketResult = (ListBucketResult)(new XmlSerializer(typeof(ListBucketResult)).Deserialize(stream));
}
XDocument root = XDocument.Parse(response.Content);
var items = (from c in root.Root.Descendants("{http://s3.amazonaws.com/doc/2006-03-01/}Contents")
select new Item()
{
Key = c.Element("{http://s3.amazonaws.com/doc/2006-03-01/}Key").Value,
LastModified = c.Element("{http://s3.amazonaws.com/doc/2006-03-01/}LastModified").Value,
ETag = c.Element("{http://s3.amazonaws.com/doc/2006-03-01/}ETag").Value,
Size = UInt64.Parse(c.Element("{http://s3.amazonaws.com/doc/2006-03-01/}Size").Value, CultureInfo.CurrentCulture),
IsDir = false
});
var prefixes = (from c in root.Root.Descendants("{http://s3.amazonaws.com/doc/2006-03-01/}CommonPrefixes")
select new Item()
{
Key = c.Element("{http://s3.amazonaws.com/doc/2006-03-01/}Prefix").Value,
IsDir = true
});
items = items.Concat(prefixes);
return new Tuple<ListBucketResult, List<Item>>(listBucketResult, items.ToList());
}
/// <summary>
/// Returns current policy stored on the server for this bucket
/// </summary>
/// <param name="bucketName">Bucket name.</param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>Task that returns the Bucket policy as a json string</returns>
public async Task<String> GetPolicyAsync(string bucketName, CancellationToken cancellationToken = default(CancellationToken))
{
IRestResponse response = null;
var path = bucketName + "?policy";
var request = await this.CreateRequest(Method.GET, bucketName,
contentType: "application/json",
resourcePath: "?policy")
.ConfigureAwait(false);
string policyString = null;
response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
var contentBytes = System.Text.Encoding.UTF8.GetBytes(response.Content);
using (var stream = new MemoryStream(contentBytes))
{
policyString = new StreamReader(stream).ReadToEnd();
}
return policyString;
}
/// <summary>
/// Sets the current bucket policy
/// </summary>
/// <param name="bucketName">Bucket Name</param>
/// <param name="policyJson">Policy json as string </param>
/// <param name="cancellationToken">Optional cancellation token to cancel the operation</param>
/// <returns>Task to set a policy</returns>
public async Task SetPolicyAsync(String bucketName, String policyJson, CancellationToken cancellationToken = default(CancellationToken))
{
var request = await this.CreateRequest(Method.PUT, bucketName,
resourcePath: "?policy",
contentType: "application/json",
body: policyJson)
.ConfigureAwait(false);
IRestResponse response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// Gets notification configuration for this bucket
/// </summary>
/// <param name="bucketName"> bucket name</param>
/// <param name="cancellationToken"> Optional cancellation token</param>
/// <returns></returns>
public async Task<BucketNotification> GetBucketNotificationsAsync(string bucketName, CancellationToken cancellationToken = default(CancellationToken))
{
utils.validateBucketName(bucketName);
var request = await this.CreateRequest(Method.GET,
bucketName,
resourcePath: "?notification")
.ConfigureAwait(false);
BucketNotification notification = null;
var response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
var contentBytes = System.Text.Encoding.UTF8.GetBytes(response.Content);
using (var stream = new MemoryStream(contentBytes))
{
notification = (BucketNotification)(new XmlSerializer(typeof(BucketNotification)).Deserialize(stream));
return notification;
}
}
/// <summary>
/// Sets the notification configuration for this bucket
/// </summary>
/// <param name="bucketName"> bucket name</param>
/// <param name="notification">notification object with configuration to be set on the server</param>
/// <param name="cancellationToken"> Optional cancellation token</param>
/// <returns></returns>
public async Task SetBucketNotificationsAsync(string bucketName, BucketNotification notification, CancellationToken cancellationToken = default(CancellationToken))
{
utils.validateBucketName(bucketName);
var request = await this.CreateRequest(Method.PUT, bucketName,
resourcePath: "?notification")
.ConfigureAwait(false);
request.XmlSerializer = new RestSharp.Serializers.DotNetXmlSerializer();
request.RequestFormat = DataFormat.Xml;
request.AddBody(notification);
IRestResponse response = await this.ExecuteTaskAsync(this.NoErrorHandlers, request, cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// Removes all bucket notification configurations stored on the server.
/// </summary>
/// <param name="bucketName"> bucket name </param>
/// <param name="cancellationToken"> optional cancellation token</param>
/// <returns></returns>
public Task RemoveAllBucketNotificationsAsync(string bucketName, CancellationToken cancellationToken = default(CancellationToken))
{
utils.validateBucketName(bucketName);
BucketNotification notification = new BucketNotification();
return SetBucketNotificationsAsync(bucketName, notification, cancellationToken);
}
}
}