[core] Refactor serverutil in preparation for eventstream

core/server.go

@@ -66,8 +66,8 @@ func NewServer(state *globalState) *grpc.Server {
 	s := grpc.NewServer()
 	grpc_health_v1.RegisterHealthServer(s, health.NewServer())
 	pb.RegisterControlServer(s, &RpcServer{
-		state:   state,
-		streams: newSafeStreamsMap(),
+		state:      state,
+		envStreams: newSafeStreamsMap(),
 	})
 	// Register reflection service on gRPC server.
 	reflection.Register(s)
@@ -112,8 +112,8 @@ func (m *RpcServer) logMethodHandled() {
 
 // Implements interface pb.ControlServer
 type RpcServer struct {
-	state   *globalState
-	streams SafeStreamsMap
+	state      *globalState
+	envStreams SafeStreamsMap
 }
 
 func (m *RpcServer) GetIntegratedServices(ctx context.Context, empty *pb.Empty) (*pb.ListIntegratedServicesReply, error) {
@@ -1064,14 +1064,14 @@ func (m *RpcServer) Subscribe(req *pb.SubscribeRequest, srv pb.Control_Subscribe
 	defer m.logMethodHandled()
 
 	for {
-		ch, ok := m.streams.GetChannel(req.GetId())
+		ch, ok := m.envStreams.GetChannel(req.GetId())
 		if !ok {
 			continue
 		}
 		select {
 		case event, ok := <-ch:
 			if !ok {
-				m.streams.delete(req.GetId())
+				m.envStreams.delete(req.GetId())
 				return nil
 			}
 			err := srv.Send(event)
@@ -1090,7 +1090,7 @@ func (m *RpcServer) NewAutoEnvironment(cxt context.Context, request *pb.NewAutoE
 	defer m.logMethodHandled()
 
 	ch := make(chan *pb.Event)
-	m.streams.add(request.GetId(), ch)
+	m.envStreams.add(request.GetId(), ch)
 	sub := environment.SubscribeToStream(ch)
 	go m.state.environments.CreateAutoEnvironment(request.GetWorkflowTemplate(), request.GetVars(), sub)
 	r := &pb.NewAutoEnvironmentReply{}

core/serverutil.go

@@ -26,7 +26,6 @@ package core
 
 import (
 	"fmt"
-	"sync"
 	"unicode/utf8"
 
 	"github.com/AliceO2Group/Control/core/repos"
@@ -190,40 +189,6 @@ func workflowToRoleTree(root workflow.Role) (ri *pb.RoleInfo) {
 	return
 }
 
-// SafeStreamsMap is a safe map where the key is usually a
-// subscriptionID received from the grpc call and as a value
-// a channel where get events from the environment
-// and we stream them to the grpc client.
-type SafeStreamsMap struct {
-	mu      sync.RWMutex
-	streams map[string]chan *pb.Event
-}
-
-func (s *SafeStreamsMap) add(id string, ch chan *pb.Event) {
-	s.mu.Lock()
-	s.streams[id] = ch
-	s.mu.Unlock()
-}
-
-func (s *SafeStreamsMap) delete(id string) {
-	s.mu.Lock()
-	delete(s.streams, id)
-	s.mu.Unlock()
-}
-
-func (s *SafeStreamsMap) GetChannel(id string) (ch chan *pb.Event, ok bool) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	ch, ok = s.streams[id]
-	return
-}
-
-func newSafeStreamsMap() SafeStreamsMap {
-	return SafeStreamsMap{
-		streams: make(map[string]chan *pb.Event),
-	}
-}
-
 func VarSpecMapToPbVarSpecMap(varSpecMap map[string]repos.VarSpec) map[string]*pb.VarSpecMessage {
 	ret := make(map[string]*pb.VarSpecMessage)
 	var vsm *pb.VarSpecMessage

core/streamutil.go

@@ -0,0 +1,66 @@
+/*
+ * === This file is part of ALICE O² ===
+ *
+ * Copyright 2020-2022 CERN and copyright holders of ALICE O².
+ * Author: Miltiadis Alexis <miltiadis.alexis@cern.ch>
+ *         Teo Mrnjavac <teo.mrnjavac@cern.ch>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * In applying this license CERN does not waive the privileges and
+ * immunities granted to it by virtue of its status as an
+ * Intergovernmental Organization or submit itself to any jurisdiction.
+ */
+
+package core
+
+import (
+	"sync"
+
+	pb "github.com/AliceO2Group/Control/core/protos"
+)
+
+// SafeStreamsMap is a safe map where the key is usually a
+// subscriptionID received from the grpc call and as a value
+// a channel where get events from the environment
+// and we stream them to the grpc client.
+type SafeStreamsMap struct {
+	mu      sync.RWMutex
+	streams map[string]chan *pb.Event
+}
+
+func (s *SafeStreamsMap) add(id string, ch chan *pb.Event) {
+	s.mu.Lock()
+	s.streams[id] = ch
+	s.mu.Unlock()
+}
+
+func (s *SafeStreamsMap) delete(id string) {
+	s.mu.Lock()
+	delete(s.streams, id)
+	s.mu.Unlock()
+}
+
+func (s *SafeStreamsMap) GetChannel(id string) (ch chan *pb.Event, ok bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	ch, ok = s.streams[id]
+	return
+}
+
+func newSafeStreamsMap() SafeStreamsMap {
+	return SafeStreamsMap{
+		streams: make(map[string]chan *pb.Event),
+	}
+}