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

watcher to support empty response (#313)

Browse Source 
* Add a WatchAsync method.

* proposal for watcher async

* fix and add testcase for async watch

* try fix build
This commit is contained in:
Boshi Lian
2019-10-22 16:02:13 -07:00
committed by Kubernetes Prow Robot
parent 8ab95b6b92
commit a95400bd50
5 changed files with 82 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
examples/watch/Program.cs
View File

@@ -13,8 +13,8 @@ namespace watch
IKubernetes client = new Kubernetes(config);
var podlistResp = client.ListNamespacedPodWithHttpMessagesAsync("default", watch: true).Result;
using (podlistResp.Watch<V1Pod>((type, item) =>
var podlistResp = client.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
using (podlistResp.Watch<V1Pod, V1PodList>((type, item) =>
{
Console.WriteLine("==on watch event==");
Console.WriteLine(type);

8
src/KubernetesClient/Kubernetes.Watch.cs
View File

@@ -149,10 +149,12 @@ namespace k8s
throw ex;
}
var stream = await httpResponse.Content.ReadAsStreamAsync().ConfigureAwait(false);
StreamReader reader = new StreamReader(stream);
return new Watcher<T>(async () => {
var stream = await httpResponse.Content.ReadAsStreamAsync().ConfigureAwait(false);
StreamReader reader = new StreamReader(stream);
return new Watcher<T>(reader, onEvent, onError, onClosed);
return reader;
}, onEvent, onError, onClosed);
}
}
}

35
src/KubernetesClient/Watcher.cs
View File

@@ -29,7 +29,9 @@ namespace k8s
public bool Watching { get; private set; }
private readonly CancellationTokenSource _cts;
private readonly StreamReader _streamReader;
private readonly Func<Task<StreamReader>> _streamReaderCreator;
private StreamReader _streamReader;
private readonly Task _watcherLoop;
/// <summary>
@@ -47,9 +49,9 @@ namespace k8s
/// <param name="onClosed">
/// The action to invoke when the server closes the connection.
/// </param>
public Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, Action<Exception> onError, Action onClosed = null)
public Watcher(Func<Task<StreamReader>> streamReaderCreator, Action<WatchEventType, T> onEvent, Action<Exception> onError, Action onClosed = null)
{
_streamReader = streamReader;
_streamReaderCreator = streamReaderCreator;
OnEvent += onEvent;
OnError += onError;
OnClosed += onClosed;
@@ -62,7 +64,7 @@ namespace k8s
public void Dispose()
{
_cts.Cancel();
_streamReader.Dispose();
_streamReader?.Dispose();
}
/// <summary>
@@ -96,9 +98,10 @@ namespace k8s
{
Watching = true;
string line;
_streamReader = await _streamReaderCreator();
// ReadLineAsync will return null when we've reached the end of the stream.
while ((line = await this._streamReader.ReadLineAsync().ConfigureAwait(false)) != null)
while ((line = await _streamReader.ReadLineAsync().ConfigureAwait(false)) != null)
{
if (cancellationToken.IsCancellationRequested)
{
@@ -147,6 +150,7 @@ namespace k8s
/// create a watch object from a call to api server with watch=true
/// </summary>
/// <typeparam name="T">type of the event object</typeparam>
/// <typeparam name="L">type of the HttpOperationResponse object</typeparam>
/// <param name="response">the api response</param>
/// <param name="onEvent">a callback when any event raised from api server</param>
/// <param name="onError">a callbak when any exception was caught during watching</param>
@@ -154,23 +158,28 @@ namespace k8s
/// The action to invoke when the server closes the connection.
/// </param>
/// <returns>a watch object</returns>
public static Watcher<T> Watch<T>(this HttpOperationResponse response,
public static Watcher<T> Watch<T, L>(this Task<HttpOperationResponse<L>> responseTask,
Action<WatchEventType, T> onEvent,
Action<Exception> onError = null,
Action onClosed = null)
{
if (!(response.Response.Content is WatcherDelegatingHandler.LineSeparatedHttpContent content))
{
throw new KubernetesClientException("not a watchable request or failed response");
}
return new Watcher<T>(async () => {
var response = await responseTask;
return new Watcher<T>(content.StreamReader, onEvent, onError, onClosed);
if (!(response.Response.Content is WatcherDelegatingHandler.LineSeparatedHttpContent content))
{
throw new KubernetesClientException("not a watchable request or failed response");
}
return content.StreamReader;
} , onEvent, onError, onClosed);
}
/// <summary>
/// create a watch object from a call to api server with watch=true
/// </summary>
/// <typeparam name="T">type of the event object</typeparam>
/// <typeparam name="L">type of the HttpOperationResponse object</typeparam>
/// <param name="response">the api response</param>
/// <param name="onEvent">a callback when any event raised from api server</param>
/// <param name="onError">a callbak when any exception was caught during watching</param>
@@ -178,12 +187,12 @@ namespace k8s
/// The action to invoke when the server closes the connection.
/// </param>
/// <returns>a watch object</returns>
public static Watcher<T> Watch<T>(this HttpOperationResponse<T> response,
public static Watcher<T> Watch<T, L>(this HttpOperationResponse<L> response,
Action<WatchEventType, T> onEvent,
Action<Exception> onError = null,
Action onClosed = null)
{
return Watch((HttpOperationResponse)response, onEvent, onError, onClosed);
return Watch(Task.FromResult(response), onEvent, onError, onClosed);
}
}
}

62
tests/KubernetesClient.Tests/WatchTests.cs
View File

@@ -66,11 +66,16 @@ namespace k8s.Tests
});
// did not pass watch param
var listTask = await client.ListNamespacedPodWithHttpMessagesAsync("default");
Assert.ThrowsAny<KubernetesClientException>(() =>
{
listTask.Watch<V1Pod>((type, item) => { });
});
var listTask = client.ListNamespacedPodWithHttpMessagesAsync("default");
var onErrorCalled = false;
using (listTask.Watch<V1Pod, V1PodList>((type, item) => { }, e => {
onErrorCalled = true;
})) { }
await Task.Delay(TimeSpan.FromSeconds(1)); // delay for onerror to be called
Assert.True(onErrorCalled);
// server did not response line by line
await Assert.ThrowsAnyAsync<Exception>(() =>
@@ -83,6 +88,41 @@ namespace k8s.Tests
}
}
[Fact]
public async Task AsyncWatcher()
{
var created = new AsyncManualResetEvent(false);
var eventsReceived = new AsyncManualResetEvent(false);
using (var server = new MockKubeApiServer(testOutput, async httpContext =>
{
// block until reponse watcher obj created
await created.WaitAsync();
await WriteStreamLine(httpContext, MockAddedEventStreamLine);
return false;
}))
{
var client = new Kubernetes(new KubernetesClientConfiguration
{
Host = server.Uri.ToString()
});
var listTask = client.ListNamespacedPodWithHttpMessagesAsync("default", watch: true);
using (listTask.Watch<V1Pod, V1PodList>((type, item) => {
eventsReceived.Set();
}))
{
// here watcher is ready to use, but http server has not responsed yet.
created.Set();
await Task.WhenAny(eventsReceived.WaitAsync(), Task.Delay(TestTimeout));
}
Assert.True(eventsReceived.IsSet);
Assert.True(created.IsSet);
}
}
[Fact]
public async Task SuriveBadLine()
{
@@ -119,7 +159,7 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod>(
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
@@ -188,7 +228,7 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var watcher = listTask.Watch<V1Pod>(
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
{
events.Add(type);
@@ -256,7 +296,7 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod>(
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
@@ -330,7 +370,7 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var errors = 0;
var watcher = listTask.Watch<V1Pod>(
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
{
testOutput.WriteLine($"Watcher received '{type}' event.");
@@ -396,7 +436,7 @@ namespace k8s.Tests
waitForException.Set();
Watcher<V1Pod> watcher;
watcher = listTask.Watch<V1Pod>(
watcher = listTask.Watch<V1Pod, V1PodList>(
onEvent: (type, item) => { },
onError: e =>
{
@@ -463,7 +503,7 @@ namespace k8s.Tests
var events = new HashSet<WatchEventType>();
var watcher = listTask.Watch<V1Pod>(
var watcher = listTask.Watch<V1Pod, V1PodList>(
(type, item) =>
{
events.Add(type);

3
tests/KubernetesClient.Tests/WatcherTests.cs
View File

@@ -3,6 +3,7 @@ using System;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Xunit;
namespace k8s.Tests
@@ -21,7 +22,7 @@ namespace k8s.Tests
ManualResetEvent mre = new ManualResetEvent(false);
Watcher<V1Pod> watcher = new Watcher<V1Pod>(
reader,
() => Task.FromResult(reader),
null,
(exception) =>
{