feat: enhance Kubernetes client with watch functionality

examples/clientset/Program.cs

@@ -12,15 +12,21 @@ private static async Task Main(string[] args)
             var config = KubernetesClientConfiguration.BuildConfigFromConfigFile();
             var client = new Kubernetes(config);
 
-            ClientSet clientSet = new ClientSet(client);
+            var clientSet = new ClientSet(client);
             var list = await clientSet.CoreV1.Pod.ListAsync("default").ConfigureAwait(false);
             foreach (var item in list)
             {
                 System.Console.WriteLine(item.Metadata.Name);
             }
 
-            var pod = await clientSet.CoreV1.Pod.GetAsync("test","default").ConfigureAwait(false);
+            var pod = await clientSet.CoreV1.Pod.GetAsync("test", "default").ConfigureAwait(false);
             System.Console.WriteLine(pod?.Metadata?.Name);
+
+            var watch = clientSet.CoreV1.Pod.WatchListAsync("default");
+            await foreach (var (_, item)in watch.ConfigureAwait(false))
+            {
+                System.Console.WriteLine(item.Metadata.Name);
+            }
         }
     }
 }

examples/watch/Program.cs

@@ -1,5 +1,4 @@
 using k8s;
-using k8s.Models;
 using System;
 using System.Threading;
 using System.Threading.Tasks;
@@ -8,9 +7,10 @@
 
 IKubernetes client = new Kubernetes(config);
 
-var podlistResp = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
+var podlistResp = client.CoreV1.WatchListNamespacedPodAsync("default");
+
 // C# 8 required https://docs.microsoft.com/en-us/archive/msdn-magazine/2019/november/csharp-iterating-with-async-enumerables-in-csharp-8
-await foreach (var (type, item) in podlistResp.WatchAsync<V1Pod, V1PodList>().ConfigureAwait(false))
+await foreach (var (type, item) in podlistResp.ConfigureAwait(false))
 {
     Console.WriteLine("==on watch event==");
     Console.WriteLine(type);
@@ -22,14 +22,24 @@
 void WatchUsingCallback(IKubernetes client)
 #pragma warning restore CS8321 // Remove unused private members
 {
-    var podlistResp = client.CoreV1.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
-    using (podlistResp.Watch<V1Pod, V1PodList>((type, item) =>
+    using var podlistResp = client.CoreV1.WatchListNamespacedPod("default");
+    podlistResp.OnEvent += (type, item) =>
     {
         Console.WriteLine("==on watch event==");
         Console.WriteLine(type);
         Console.WriteLine(item.Metadata.Name);
         Console.WriteLine("==on watch event==");
-    }))
+    };
+    podlistResp.OnError += (error) =>
+    {
+        Console.WriteLine("==on watch error==");
+        Console.WriteLine(error.Message);
+        Console.WriteLine("==on watch error==");
+    };
+    podlistResp.OnClosed += () =>
+    {
+        Console.WriteLine("==on watch closed==");
+    };
     {
         Console.WriteLine("press ctrl + c to stop watching");
 

src/KubernetesClient.Aot/KubernetesClient.Aot.csproj

@@ -89,8 +89,8 @@
     <Compile Include="..\KubernetesClient\IKubernetes.Exec.cs" />
     <Compile Include="..\KubernetesClient\Kubernetes.Exec.cs" />
 
-    <!-- <Compile Include="..\KubernetesClient\Watcher.cs" /> -->
-    <!-- <Compile Include="..\KubernetesClient\WatcherExt.cs" /> -->
+    <Compile Include="..\KubernetesClient\Watcher.cs" />
+    <Compile Include="..\KubernetesClient\WatcherExt.cs" />
     <Compile Include="..\KubernetesClient\LineSeparatedHttpContent.cs" />
 
     <Compile Include="..\KubernetesClient\Exceptions\KubeConfigException.cs" />

src/LibKubernetesGenerator/ParamHelper.cs

@@ -3,6 +3,7 @@
 using Scriban.Runtime;
 using System;
 using System.Linq;
+using System.Collections.Generic;
 
 namespace LibKubernetesGenerator
 {
@@ -21,6 +22,8 @@ public void RegisterHelper(ScriptObject scriptObject)
         {
             scriptObject.Import(nameof(GetModelCtorParam), new Func<JsonSchema, string>(GetModelCtorParam));
             scriptObject.Import(nameof(IfParamContains), IfParamContains);
+            scriptObject.Import(nameof(FilterParameters), FilterParameters);
+            scriptObject.Import(nameof(GetParameterValueForWatch), new Func<OpenApiParameter, bool, string, string>(GetParameterValueForWatch));
         }
 
         public static bool IfParamContains(OpenApiOperation operation, string name)
@@ -39,6 +42,23 @@ public static bool IfParamContains(OpenApiOperation operation, string name)
             return found;
         }
 
+        public static IEnumerable<OpenApiParameter> FilterParameters(OpenApiOperation operation, string excludeParam)
+        {
+            return operation.Parameters.Where(p => p.Name != excludeParam);
+        }
+
+        public string GetParameterValueForWatch(OpenApiParameter parameter, bool watch, string init = "false")
+        {
+            if (parameter.Name == "watch")
+            {
+                return watch ? "true" : "false";
+            }
+            else
+            {
+                return generalNameHelper.GetDotNetNameOpenApiParameter(parameter, init);
+            }
+        }
+
         public string GetModelCtorParam(JsonSchema schema)
         {
             return string.Join(", ", schema.Properties.Values

src/LibKubernetesGenerator/TypeHelper.cs

@@ -245,6 +245,14 @@ string toType()
                     }
 
                     break;
+                case "T":
+                    // Return single item type from list type (e.g., V1Pod from V1PodList)
+                    return !string.IsNullOrEmpty(t) && t.EndsWith("List", StringComparison.Ordinal)
+                        ? t.Substring(0, t.Length - 4)
+                        : t;
+                case "TList":
+                    // Return list type as-is
+                    return t;
             }
 
             return t;

src/LibKubernetesGenerator/templates/Client.cs.template

@@ -17,10 +17,11 @@ public partial class {{name}}Client : ResourceClient
     }
 
     {{for api in apis }}
+    {{~ $filteredParams = FilterParameters api.operation "watch" ~}}
     /// <summary>
     /// {{ToXmlDoc api.operation.description}}
     /// </summary>
-    {{ for parameter in api.operation.parameters}}
+    {{ for parameter in $filteredParams}}
     /// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
     /// {{ToXmlDoc parameter.description}}
     /// </param>
@@ -29,15 +30,15 @@ public partial class {{name}}Client : ResourceClient
     /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous operation.
     /// </param>
     public async Task{{GetReturnType api.operation "<>"}} {{GetActionName api.operation name "Async"}}(
-        {{ for parameter in api.operation.parameters}}
+        {{ for parameter in $filteredParams}}
         {{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
         {{ end }}
         CancellationToken cancellationToken = default(CancellationToken))
     {
         {{if IfReturnType api.operation "stream"}}
         var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
             {{ for parameter in api.operation.parameters}}
-            {{GetDotNetNameOpenApiParameter parameter "false"}},
+            {{GetParameterValueForWatch parameter false}},
             {{end}}
             null,
             cancellationToken);
@@ -47,7 +48,7 @@ public partial class {{name}}Client : ResourceClient
         {{if IfReturnType api.operation "obj"}}
         using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
             {{ for parameter in api.operation.parameters}}
-            {{GetDotNetNameOpenApiParameter parameter "false"}},
+            {{GetParameterValueForWatch parameter false}},
             {{end}}
             null,
             cancellationToken).ConfigureAwait(false))
@@ -58,7 +59,7 @@ public partial class {{name}}Client : ResourceClient
         {{if IfReturnType api.operation "void"}}
         using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
             {{ for parameter in api.operation.parameters}}
-            {{GetDotNetNameOpenApiParameter parameter "false"}},
+            {{GetParameterValueForWatch parameter false}},
             {{end}}
             null,
             cancellationToken).ConfigureAwait(false))
@@ -71,7 +72,7 @@ public partial class {{name}}Client : ResourceClient
     /// <summary>
     /// {{ToXmlDoc api.operation.description}}
     /// </summary>
-    {{ for parameter in api.operation.parameters}}
+    {{ for parameter in $filteredParams}}
     /// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
     /// {{ToXmlDoc parameter.description}}
     /// </param>
@@ -80,14 +81,14 @@ public partial class {{name}}Client : ResourceClient
     /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous operation.
     /// </param>
     public async Task<T> {{GetActionName api.operation name "Async"}}<T>(
-        {{ for parameter in api.operation.parameters}}
+        {{ for parameter in $filteredParams}}
         {{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "false"}},
         {{ end }}
         CancellationToken cancellationToken = default(CancellationToken))
     {
         using (var _result = await Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}<T>(
             {{ for parameter in api.operation.parameters}}
-            {{GetDotNetNameOpenApiParameter parameter "false"}},
+            {{GetParameterValueForWatch parameter false}},
             {{end}}
             null,
             cancellationToken).ConfigureAwait(false))
@@ -96,5 +97,68 @@ public partial class {{name}}Client : ResourceClient
         }
     }
     {{end}}
+
+    {{if IfParamContains api.operation "watch"}}
+    /// <summary>
+    /// Watch {{ToXmlDoc api.operation.description}}
+    /// </summary>
+    {{ for parameter in $filteredParams}}
+    /// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
+    /// {{ToXmlDoc parameter.description}}
+    /// </param>
+    {{ end }}
+    /// <param name="onEvent">Callback when any event raised from api server</param>
+    /// <param name="onError">Callback when any exception was caught during watching</param>
+    /// <param name="onClosed">Callback when the server closes the connection</param>
+    public Watcher<{{GetReturnType api.operation "T"}}> Watch{{GetActionName api.operation name ""}}(
+        {{ for parameter in $filteredParams}}
+        {{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
+        {{ end }}
+        Action<WatchEventType, {{GetReturnType api.operation "T"}}> onEvent = null,
+        Action<Exception> onError = null,
+        Action onClosed = null)
+    {
+        if (onEvent == null) throw new ArgumentNullException(nameof(onEvent));
+
+        var responseTask = Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
+            {{ for parameter in api.operation.parameters}}
+            {{GetParameterValueForWatch parameter true}},
+            {{ end }}
+            null,
+            CancellationToken.None);
+
+        return responseTask.Watch<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
+            onEvent, onError, onClosed);
+    }
+
+    /// <summary>
+    /// Watch {{ToXmlDoc api.operation.description}} as async enumerable
+    /// </summary>
+    {{ for parameter in $filteredParams}}
+    /// <param name="{{GetDotNetNameOpenApiParameter parameter "false"}}">
+    /// {{ToXmlDoc parameter.description}}
+    /// </param>
+    {{ end }}
+    /// <param name="onError">Callback when any exception was caught during watching</param>
+    /// <param name="cancellationToken">Cancellation token</param>
+    public IAsyncEnumerable<(WatchEventType, {{GetReturnType api.operation "T"}})> Watch{{GetActionName api.operation name "Async"}}(
+        {{ for parameter in $filteredParams}}
+        {{GetDotNetTypeOpenApiParameter parameter}} {{GetDotNetNameOpenApiParameter parameter "true"}},
+        {{ end }}
+        Action<Exception> onError = null,
+        CancellationToken cancellationToken = default)
+    {
+        var responseTask = Client.{{group}}.{{GetOperationId api.operation "WithHttpMessagesAsync"}}(
+            {{ for parameter in api.operation.parameters}}
+            {{GetParameterValueForWatch parameter true}},
+            {{ end }}
+            null,
+            cancellationToken);
+
+        return responseTask.WatchAsync<{{GetReturnType api.operation "T"}}, {{GetReturnType api.operation "TList"}}>(
+            onError, cancellationToken);
+    }
     {{end}}
+
+    {{end}} 
 }