feat: add drawio

internal/crd/watch.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"gitea.com/logicamp/lc"
+	"github.com/reactivex/rxgo/v2"
 	"go.uber.org/zap"
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -13,40 +14,53 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 )
 
-func (crd Crd) watchFlinkJobs() {
-	lc.Logger.Debug("[crd] starting watch")
-	watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
-	if err != nil {
-		panic(err)
-	}
-	defer watcher.Stop()
+func (crd Crd) watchFlinkJobs() rxgo.Observable {
 
-	for event := range watcher.ResultChan() {
-		lc.Logger.Debug("[crd] event received", zap.Any("type", event.Type))
-		unstructuredJob := event.Object.(*unstructured.Unstructured)
-		unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
-		if err != nil {
-			lc.Logger.Error("cannot create unstructured map", zap.Error(err))
-			continue
-		}
-		job := &v1alpha1.FlinkJob{}
-		err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
-		if err != nil {
-			lc.Logger.Error("cannot convert unstructured to structured", zap.Error(err))
-			continue
-		}
-		switch event.Type {
-		case watch.Bookmark:
-		case watch.Modified:
-			fmt.Printf("FlinkJob updated: %s\n", job.GetName())
-			// Handle the new FlinkJob
-			crd.repsert(job)
-		case watch.Added:
-			fmt.Printf("New FlinkJob created: %s\n", job.GetName())
-			// Handle the new FlinkJob
-			crd.repsert(job)
-		case watch.Deleted:
+	ch := make(chan rxgo.Item)
 
+	go func() {
+		lc.Logger.Debug("[crd] starting watch")
+		watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
+		if err != nil {
+			panic(err)
 		}
-	}
+		defer watcher.Stop()
+		for event := range watcher.ResultChan() {
+			lc.Logger.Debug("[crd] event received", zap.Any("type", event.Type))
+			unstructuredJob := event.Object.(*unstructured.Unstructured)
+			unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
+			if err != nil {
+				lc.Logger.Error("cannot create unstructured map", zap.Error(err))
+				continue
+			}
+			job := &v1alpha1.FlinkJob{}
+
+			err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
+			if err != nil {
+				lc.Logger.Error("cannot convert unstructured to structured", zap.Error(err))
+				continue
+			}
+
+			ch <- rxgo.Item{
+				V: &FlinkJobCrdEvent{
+					EventType: event.Type,
+					Job:       job,
+				},
+			}
+			switch event.Type {
+			case watch.Bookmark:
+			case watch.Modified:
+				lc.Logger.Info("[crd] [watch] flink job updated")
+				fmt.Printf("FlinkJob updated: %s\n", job.GetName())
+				crd.repsert(job)
+			case watch.Added:
+				lc.Logger.Info("[crd] [watch] new flink job created")
+				crd.repsert(job)
+			case watch.Deleted:
+			}
+		}
+
+	}()
+
+	return rxgo.FromChannel(ch)
 }