oc-catalog/controllers/general.go

package controllers

import (
	"context"
	"encoding/json"
	"fmt"
	"oc-catalog/infrastructure"

	oclib "cloud.o-forge.io/core/oc-lib"
	w "cloud.o-forge.io/core/oc-lib/models/workflow"
	tools "cloud.o-forge.io/core/oc-lib/tools"
	beego "github.com/beego/beego/v2/server/web"
	"github.com/gorilla/websocket"
)

// Operations about compute
type GeneralController struct {
	beego.Controller
}

// @Title GetAll
// @Description find compute by id
// @Param file formData file true "File to upload"
// @Success 200 {compute} models.workflow
// @Failure 406 {string} string "Bad request"
// @router / [get]
func (o *GeneralController) GetAll() {
	user, peerID, groups := oclib.ExtractTokenInfo(*o.Ctx.Request)
	file, _, err := o.Ctx.Request.FormFile("file")
	if err != nil {
		o.Data["json"] = map[string]interface{}{
			"data":  nil,
			"code":  406,
			"error": err.Error(),
		}
		o.ServeJSON()
		return
	}
	newWorkflow := &w.Workflow{}
	req := &tools.APIRequest{
		Username: user,
		PeerID:   peerID,
		Groups:   groups,
		Admin:    true,
	}
	newWorkflow, err = newWorkflow.ExtractFromPlantUML(file, req)
	if err != nil {
		o.Data["json"] = map[string]interface{}{
			"data":  nil,
			"code":  406,
			"error": err.Error(),
		}
		o.ServeJSON()
		return
	}

	o.Data["json"] = map[string]interface{}{
		"data":  newWorkflow,
		"code":  200,
		"error": nil,
	}

	o.ServeJSON()
}

func Websocket(ctx context.Context, user string, groups []string, dataType int, conn *websocket.Conn) {
	defer conn.Close()
	done := make(chan struct{})
	go func() {
		var discard interface{}
		for {
			if err := conn.ReadJSON(&discard); err != nil {
				close(done)
				return
			}
		}
	}()
	defer func() {
		if ch, ok := infrastructure.SearchStream[user]; ok {
			close(ch)
			infrastructure.SearchMu.Lock()
			delete(infrastructure.SearchStream, user)
			infrastructure.SearchMu.Unlock()
		}
		fmt.Println("CLOSE !")
		infrastructure.EmitNATS(user, nil, tools.PropalgationMessage{
			Action:   tools.PB_CLOSE_SEARCH,
			DataType: dataType,
		})
	}()
	for {
		select {
		case msg, ok := <-infrastructure.SearchStream[user]:
			if !ok {
				continue
			}
			m := map[string]interface{}{}
			if err := json.Unmarshal(msg, &m); err == nil {
				if err := conn.WriteJSON(m); err != nil {
					continue
				}
			} else {
				continue
			}
		case <-done:
			return
		case <-ctx.Done():
			return
		}
	}
}