Watchers should send a notification when the connection is closed

[qmfrederik]

The Kubernetes watch APIs have a timeout, after which the connection is closed.

Each watch requests gets assigned a random timeout between [timeout, 2 * timeout]. The default timeout is 30 minutes, see kubernetes/kubernetes#15705.

The Watcher updates the Watching property but doesn't raise an event/invoke a callback when the server terminates the request.

Suggestion: make the Watcher APIs take an onClosed callback, and have the Watcher class expose an Closed event.

Background/rationale: I had naïve code like this which was using a watcher to block until a job runs to completion, but it would just idle forever when the connection is closed by the server:

using (var watcher = await kubernetes.WatchObjectAsync<V1Job>(
    path: $"apis/batch/v1/watch/namespaces/{job.Metadata.NamespaceProperty}/jobs/{job.Metadata.Name}",
    resourceVersion: resourceVersion,
    onEvent: (eventType, j) =>
    {
        logger.LogInformation($"Got a {eventType} event for job {j.Metadata.Name}, resource version {j.Metadata.ResourceVersion}. The job completion time is set to {j.Status?.CompletionTime}");
        resourceVersion = j.Metadata.ResourceVersion;

        if (eventType == WatchEventType.Deleted)
        {
            logger.LogInformation($"The job was deleted.");
            jobCompleted = true;
            mre.Set();
        }
        else if (j.Status.HasCompleted())
        {
            logger.LogInformation($"The job has completed.");
            jobCompleted = true;
            mre.Set();
        }
        else if (j.Status.HasFailed())
        {
            logger.LogInformation($"The job has failed. {j.Status.GetFailureMessage()}.");
            jobCompleted = true;
            mre.Set();
        }
    },
    onError: (ex) =>
    {
        logger.LogInformation($"An error occurred within the watcher while polling. {ex.Message}");

        // Signal the calling thread so it will try again.
        mre.Set();
    },
    cancellationToken: cancellationToken).ConfigureAwait(false))
{
    await mre.WaitAsync(cancellationToken).ConfigureAwait(false);
}