[core] Allow templating of FairMQ channel rateLogging

Author: teo

core/task/channel/channel.go

@@ -40,7 +40,7 @@ type Channel struct {
 	Type        ChannelType             `yaml:"type"`
 	SndBufSize  int                     `yaml:"sndBufSize"`
 	RcvBufSize  int                     `yaml:"rcvBufSize"`
-	RateLogging int                     `yaml:"rateLogging"`
+	RateLogging string                  `yaml:"rateLogging"`//actually an int but we allow templating
 	Transport   TransportType           `yaml:"transport"`  //default: default
 }
 
@@ -82,10 +82,7 @@ func (c *Channel) UnmarshalYAML(unmarshal func(interface{}) error) (err error) {
 	if err != nil {
 		return
 	}
-	c.RateLogging, err = strconv.Atoi(aux.RateLogging)
-	if err != nil {
-		return
-	}
+	c.RateLogging = aux.RateLogging
 	c.Transport = aux.Transport
 
 	return

core/task/channel/inbound.go

@@ -67,7 +67,7 @@ func (inbound Inbound) MarshalYAML() (interface{}, error) {
 		Type        ChannelType   `yaml:"type"`
 		SndBufSize  int           `yaml:"sndBufSize,omitempty"`
 		RcvBufSize  int           `yaml:"rcvBufSize,omitempty"`
-		RateLogging int           `yaml:"rateLogging,omitempty"`
+		RateLogging string        `yaml:"rateLogging,omitempty"`
 		Transport   TransportType `yaml:"transport"`
 		Addressing  AddressFormat `yaml:"addressing"`
 	}
@@ -114,7 +114,7 @@ func (inbound *Inbound) buildFMQMap(address string, transport TransportType) (pm
 	chanProps := controlcommands.PropertyMap{
 		"address": address,
 		"method": "bind",
-		"rateLogging": strconv.Itoa(inbound.RateLogging),
+		"rateLogging": inbound.RateLogging,
 		"rcvBufSize": strconv.Itoa(inbound.RcvBufSize),
 		"rcvKernelSize": "0", //NOTE: hardcoded
 		"sndBufSize": strconv.Itoa(inbound.SndBufSize),

core/task/channel/outbound.go

@@ -65,7 +65,7 @@ func (outbound Outbound) MarshalYAML() (interface{}, error) {
 		Type        ChannelType   `yaml:"type"`
 		SndBufSize  int           `yaml:"sndBufSize,omitempty"`
 		RcvBufSize  int           `yaml:"rcvBufSize,omitempty"`
-		RateLogging int           `yaml:"rateLogging,omitempty"`
+		RateLogging string        `yaml:"rateLogging,omitempty"`
 		Transport   TransportType `yaml:"transport"`
 		Target      string        `yaml:"target"`
 	}
@@ -141,7 +141,7 @@ func (outbound *Outbound) buildFMQMap(address string, transport TransportType) (
 	chanProps := controlcommands.PropertyMap {
 		"address": address,
 		"method": "connect",
-		"rateLogging": strconv.Itoa(outbound.RateLogging),
+		"rateLogging": outbound.RateLogging,
 		"rcvBufSize": strconv.Itoa(outbound.RcvBufSize),
 		"rcvKernelSize": "0", //NOTE: hardcoded
 		"sndBufSize": strconv.Itoa(outbound.SndBufSize),

core/task/task.go

@@ -469,6 +469,41 @@ func (t *Task) BuildPropertyMap(bindMap channel.BindMap) (propMap controlcommand
 				propMap[k] = v
 			}
 
+			// For FAIRMQ tasks, we append FairMQ channel configuration
+			if class.Control.Mode == controlmode.FAIRMQ ||
+				class.Control.Mode == controlmode.DIRECT {
+				for _, inbCh := range channel.MergeInbound(t.GetParent().CollectInboundChannels(), class.Bind) {
+					endpoint, ok := t.localBindMap[inbCh.Name]
+					if !ok {
+						log.WithFields(logrus.Fields{
+							"channelName": inbCh.Name,
+							"taskName": t.name,
+						}).
+							Error("endpoint not allocated for inbound channel")
+						continue
+					}
+
+					// We get the FairMQ-formatted propertyMap from the inbound channel spec
+					chanProps := inbCh.ToFMQMap(endpoint)
+
+					// And we copy it into the task's propertyMap
+					for k, v := range chanProps {
+						propMap[k] = v
+					}
+				}
+				for _, outboundCh := range channel.MergeOutbound(t.GetParent().CollectOutboundChannels(), class.Connect) {
+					// We get the FairMQ-formatted propertyMap from the outbound channel spec
+					chanProps := outboundCh.ToFMQMap(bindMap)
+
+					// And if valid, we copy it into the task's propertyMap
+					if len(chanProps) > 0 {
+						for k, v := range chanProps {
+							propMap[k] = v
+						}
+					}
+				}
+			} // end append FairMQ configuration
+
 			objStack := make(map[string]interface{})
 			objStack["ToPtree"] = template.MakeToPtreeFunc(varStack, propMap)
 
@@ -503,40 +538,6 @@ func (t *Task) BuildPropertyMap(bindMap channel.BindMap) (propMap controlcommand
 			}
 		}
 
-		// For FAIRMQ tasks, we append FairMQ channel configuration
-		if class.Control.Mode == controlmode.FAIRMQ ||
-			class.Control.Mode == controlmode.DIRECT {
-			for _, inbCh := range channel.MergeInbound(t.GetParent().CollectInboundChannels(), class.Bind) {
-				endpoint, ok := t.localBindMap[inbCh.Name]
-				if !ok {
-					log.WithFields(logrus.Fields{
-							"channelName": inbCh.Name,
-							"taskName": t.name,
-						}).
-						Error("endpoint not allocated for inbound channel")
-					continue
-				}
-
-				// We get the FairMQ-formatted propertyMap from the inbound channel spec
-				chanProps := inbCh.ToFMQMap(endpoint)
-
-				// And we copy it into the task's propertyMap
-				for k, v := range chanProps {
-					propMap[k] = v
-				}
-			}
-			for _, outboundCh := range channel.MergeOutbound(t.GetParent().CollectOutboundChannels(), class.Connect) {
-				// We get the FairMQ-formatted propertyMap from the outbound channel spec
-				chanProps := outboundCh.ToFMQMap(bindMap)
-
-				// And if valid, we copy it into the task's propertyMap
-				if len(chanProps) > 0 {
-					for k, v := range chanProps {
-						propMap[k] = v
-					}
-				}
-			}
-		}
 	}
 	return propMap
 }