[executor] Support switching between Protobuf and JSON transport

Author: teo

executor/executable/controllabletask.go

@@ -33,6 +33,7 @@ import (
 	"time"
 
 	"github.com/AliceO2Group/Control/common/event"
+	"github.com/AliceO2Group/Control/common/utils"
 	"github.com/AliceO2Group/Control/core/controlcommands"
 	"github.com/AliceO2Group/Control/executor/executorcmd"
 	pb "github.com/AliceO2Group/Control/executor/protos"
@@ -94,7 +95,12 @@ func (t *ControllableTask) Launch() error {
 		"id":          t.ti.TaskID.Value,
 	}).
 	Debug("starting gRPC client")
-	t.rpc = executorcmd.NewClient(t.tci.ControlPort, t.tci.ControlMode)
+
+	controlTransport := executorcmd.ProtobufTransport
+	if utils.StringSliceContains(taskCmd.Args, "OCClite") {
+		controlTransport = executorcmd.JsonTransport
+	}
+	t.rpc = executorcmd.NewClient(t.tci.ControlPort, t.tci.ControlMode, controlTransport)
 	if t.rpc == nil {
 		return errors.New("could not start gRPC client")
 	}

executor/executorcmd/client.go

@@ -48,12 +48,18 @@ import (
 
 var log = logger.New(logrus.StandardLogger(), "executorcmd")
 
-func NewClient(controlPort uint64, controlMode controlmode.ControlMode) *RpcClient {
+type ControlTransport uint32
+const (
+	ProtobufTransport = ControlTransport(0)
+	JsonTransport = ControlTransport(1)
+)
+
+func NewClient(controlPort uint64, controlMode controlmode.ControlMode, controlTransport ControlTransport) *RpcClient {
 	endpoint := fmt.Sprintf("127.0.0.1:%d", controlPort)
 	log.WithField("endpoint", endpoint).Debug("starting new gRPC client")
 
 	cxt, cancel := context.WithTimeout(context.Background(), 20*time.Second)
-	conn, err := grpc.DialContext(cxt, endpoint, grpc.WithInsecure())
+	conn, err := grpc.DialContext(cxt, endpoint, grpc.WithInsecure(), grpc.WithBlock())
 	if err != nil {
 		log.WithField("error", err.Error()).
 			WithField("endpoint", endpoint).
@@ -62,9 +68,15 @@ func NewClient(controlPort uint64, controlMode controlmode.ControlMode) *RpcClie
 		return nil
 	}
 
+	var occClient pb.OccClient
+	if controlTransport == JsonTransport {
+		occClient = nopb.NewOccClient(conn)
+	} else {
+		occClient = pb.NewOccClient(conn)
+	}
+
 	client := &RpcClient {
-		//OccClient: pb.NewOccClient(conn),
-		OccClient: nopb.NewOccClient(conn),
+		OccClient: occClient,
 		conn: conn,
 	}
 

occ/peanut/peanut.go

@@ -280,7 +280,7 @@ func Run(cmdString string) (err error) {
 	// Setup RPC
 	go func() {
 		// FIXME allow choice of controlmode.FAIRMQ
-		rpcClient = executorcmd.NewClient(occPort, controlmode.DIRECT)
+		rpcClient = executorcmd.NewClient(occPort, controlmode.DIRECT, executorcmd.ProtobufTransport)
 		var response *pb.GetStateReply
 		response, err = rpcClient.GetState(context.TODO(), &pb.GetStateRequest{}, grpc.EmptyCallOption{})
 		if err != nil {