kubernetes-client/csharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add generic watch (#1155)

Browse Source 
* add generic watch

* Update WatcherExt.cs
This commit is contained in:
Boshi Lian
2023-01-03 15:37:32 -08:00
committed by GitHub
parent 2a13d46a51
commit d38302d8f1
2 changed files with 33 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/KubernetesClient/GenericClient.cs
View File

@@ -115,6 +115,34 @@ namespace k8s
return KubernetesJson.Deserialize<T>(resp.Body.ToString()); return KubernetesJson.Deserialize<T>(resp.Body.ToString());
} }
public IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T>(Action<Exception> onError = null, CancellationToken cancel = default)
where T : IKubernetesObject
{
var respTask = kubernetes.CustomObjects.ListClusterCustomObjectWithHttpMessagesAsync(group, version, plural, watch: true, cancellationToken: cancel);
return respTask.WatchAsync<T, object>();
}
public IAsyncEnumerable<(WatchEventType, T)> WatchNamespacedAsync<T>(string ns, Action<Exception> onError = null, CancellationToken cancel = default)
where T : IKubernetesObject
{
var respTask = kubernetes.CustomObjects.ListNamespacedCustomObjectWithHttpMessagesAsync(group, version, ns, plural, watch: true, cancellationToken: cancel);
return respTask.WatchAsync<T, object>();
}
public Watcher<T> Watch<T>(Action<WatchEventType, T> onEvent, Action<Exception> onError = null, Action onClosed = null)
where T : IKubernetesObject
{
var respTask = kubernetes.CustomObjects.ListClusterCustomObjectWithHttpMessagesAsync(group, version, plural, watch: true);
return respTask.Watch(onEvent, onError, onClosed);
}
public Watcher<T> WatchNamespaced<T>(string ns, Action<WatchEventType, T> onEvent, Action<Exception> onError = null, Action onClosed = null)
where T : IKubernetesObject
{
var respTask = kubernetes.CustomObjects.ListNamespacedCustomObjectWithHttpMessagesAsync(group, version, ns, plural, watch: true);
return respTask.Watch(onEvent, onError, onClosed);
}
public void Dispose() public void Dispose()
{ {
Dispose(true); Dispose(true);

7
src/KubernetesClient/WatcherExt.cs
View File

@@ -1,6 +1,7 @@
using k8s.Autorest; using k8s.Autorest;
using k8s.Exceptions; using k8s.Exceptions;
using System.IO; using System.IO;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
namespace k8s namespace k8s
@@ -72,12 +73,14 @@ namespace k8s
/// <typeparam name="L">type of the HttpOperationResponse object</typeparam> /// <typeparam name="L">type of the HttpOperationResponse object</typeparam>
/// <param name="responseTask">the api response</param> /// <param name="responseTask">the api response</param>
/// <param name="onError">a callbak when any exception was caught during watching</param> /// <param name="onError">a callbak when any exception was caught during watching</param>
/// <param name="cancellationToken">cancellation token</param>
/// <returns>IAsyncEnumerable of watch events</returns> /// <returns>IAsyncEnumerable of watch events</returns>
public static IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T, L>( public static IAsyncEnumerable<(WatchEventType, T)> WatchAsync<T, L>(
this Task<HttpOperationResponse<L>> responseTask, this Task<HttpOperationResponse<L>> responseTask,
Action<Exception> onError = null) Action<Exception> onError = null,
CancellationToken cancellationToken = default)
{ {
return Watcher<T>.CreateWatchEventEnumerator(MakeStreamReaderCreator<T, L>(responseTask), onError); return Watcher<T>.CreateWatchEventEnumerator(MakeStreamReaderCreator<T, L>(responseTask), onError, cancellationToken);
} }
} }
} }