feat: implement WatchEvents for argoKubeWorkflowServiceClient. Fixes argoproj#6173 (argoproj#6816)

Signed-off-by: NikeNano <[email protected]>

Author: NikeNano

pkg/apiclient/argo-kube-workflow-service-client.go

@@ -2,7 +2,6 @@ package apiclient
 
 import (
 	"context"
-	"errors"
 	"io"
 
 	"google.golang.org/grpc"
@@ -30,7 +29,7 @@ func (c *argoKubeWorkflowServiceClient) ListWorkflows(ctx context.Context, req *
 }
 
 func (c *argoKubeWorkflowServiceClient) WatchWorkflows(ctx context.Context, req *workflowpkg.WatchWorkflowsRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_WatchWorkflowsClient, error) {
-	intermediary := newWatchIntermediary(ctx)
+	intermediary := newWorkflowWatchIntermediary(ctx)
 	go func() {
 		defer intermediary.cancel()
 		err := c.delegate.WatchWorkflows(req, intermediary)
@@ -43,8 +42,18 @@ func (c *argoKubeWorkflowServiceClient) WatchWorkflows(ctx context.Context, req
 	return intermediary, nil
 }
 
-func (c *argoKubeWorkflowServiceClient) WatchEvents(context.Context, *workflowpkg.WatchEventsRequest, ...grpc.CallOption) (workflowpkg.WorkflowService_WatchEventsClient, error) {
-	return nil, errors.New("not implemented")
+func (c *argoKubeWorkflowServiceClient) WatchEvents(ctx context.Context, req *workflowpkg.WatchEventsRequest, _ ...grpc.CallOption) (workflowpkg.WorkflowService_WatchEventsClient, error) {
+	intermediary := newEventWatchIntermediary(ctx)
+	go func() {
+		defer intermediary.cancel()
+		err := c.delegate.WatchEvents(req, intermediary)
+		if err != nil {
+			intermediary.error <- err
+		} else {
+			intermediary.error <- io.EOF
+		}
+	}()
+	return intermediary, nil
 }
 
 func (c *argoKubeWorkflowServiceClient) DeleteWorkflow(ctx context.Context, req *workflowpkg.WorkflowDeleteRequest, _ ...grpc.CallOption) (*workflowpkg.WorkflowDeleteResponse, error) {

pkg/apiclient/watch-intermediary.go

@@ -5,20 +5,52 @@ import (
 
 	"google.golang.org/grpc/metadata"
 
+	v1 "k8s.io/api/core/v1"
+
 	workflowpkg "github.com/argoproj/argo-workflows/v3/pkg/apiclient/workflow"
 )
 
-type watchIntermediary struct {
+type workflowWatchIntermediary struct {
 	abstractIntermediary
 	events chan *workflowpkg.WorkflowWatchEvent
 }
 
-func (w watchIntermediary) Send(e *workflowpkg.WorkflowWatchEvent) error {
+func (w workflowWatchIntermediary) Send(e *workflowpkg.WorkflowWatchEvent) error {
+	w.events <- e
+	return nil
+}
+
+func (w workflowWatchIntermediary) Recv() (*workflowpkg.WorkflowWatchEvent, error) {
+	select {
+	case e := <-w.error:
+		return nil, e
+	case event := <-w.events:
+		return event, nil
+	}
+}
+
+func (w *workflowWatchIntermediary) SendHeader(metadata.MD) error {
+	// We invoke `SendHeader` in order to eagerly flush headers to allow us to send period
+	// keepalives when using HTTP/1 and Server Sent Events, so we need to implement this here,
+	// though we don't use the meta for anything.
+	return nil
+}
+
+func newWorkflowWatchIntermediary(ctx context.Context) *workflowWatchIntermediary {
+	return &workflowWatchIntermediary{newAbstractIntermediary(ctx), make(chan *workflowpkg.WorkflowWatchEvent)}
+}
+
+type eventWatchIntermediary struct {
+	abstractIntermediary
+	events chan *v1.Event
+}
+
+func (w eventWatchIntermediary) Send(e *v1.Event) error {
 	w.events <- e
 	return nil
 }
 
-func (w watchIntermediary) Recv() (*workflowpkg.WorkflowWatchEvent, error) {
+func (w eventWatchIntermediary) Recv() (*v1.Event, error) {
 	select {
 	case e := <-w.error:
 		return nil, e
@@ -27,13 +59,13 @@ func (w watchIntermediary) Recv() (*workflowpkg.WorkflowWatchEvent, error) {
 	}
 }
 
-func (w *watchIntermediary) SendHeader(metadata.MD) error {
+func (w *eventWatchIntermediary) SendHeader(metadata.MD) error {
 	// We invoke `SendHeader` in order to eagerly flush headers to allow us to send period
 	// keepalives when using HTTP/1 and Server Sent Events, so we need to implement this here,
 	// though we don't use the meta for anything.
 	return nil
 }
 
-func newWatchIntermediary(ctx context.Context) *watchIntermediary {
-	return &watchIntermediary{newAbstractIntermediary(ctx), make(chan *workflowpkg.WorkflowWatchEvent)}
+func newEventWatchIntermediary(ctx context.Context) *eventWatchIntermediary {
+	return &eventWatchIntermediary{newAbstractIntermediary(ctx), make(chan *v1.Event)}
 }